diff --git a/docs/README.md b/docs/README.md index 40e42904..81c64bb3 100644 --- a/docs/README.md +++ b/docs/README.md @@ -12,6 +12,8 @@ Read documents at the library revision used by your consumer. These are source-l ## Runtime contracts +- [Cloud Run revision tags](cloud-run-tags.md): pinned offline allocation, exact-target warmup policy and caller restore requirements. + - [One-time setup intent](strategy-release-runtime-contract.zh-CN.md#一次确认的自动化-setup-intent): immutable, non-executing owner scope and Decisions handoff; no runtime adoption or activation authority. - [Platform notification outcomes](platform-notification-outcomes.md): credential-free intents, receipts and log redaction. diff --git a/docs/cloud-run-tags.md b/docs/cloud-run-tags.md new file mode 100644 index 00000000..ba011bb6 --- /dev/null +++ b/docs/cloud-run-tags.md @@ -0,0 +1,107 @@ +# Cloud Run revision tags + +`quant_platform_kit.operators` exports the pinned offline operator contract +`cloud-run-tags/v1`. It performs no network calls, deployment, traffic updates, +renames, IAM changes or credential access. Runtime trading controls are outside +this module. Existing callers acquire no additional authority by importing it. + +## Allocation + +Call `headroom(service_name)` on the exact original service name, before display +redaction. The tagged URL budget is `len(service_name) + len(tag) <= 46`. +`allocate_tag` takes the exact service name, preferred tag, project, region, +candidate revision, operation/run identity, a complete fresh `existing_tags` +mapping, and a private job-local `state_path`. + +- A fitting preferred tag is selected only when free or bound to this exact + operation and candidate in the journal. A cloud revision match alone does + not prove ownership. Foreign or unknown occupancy returns `TAG_OCCUPIED`. + The first occupancy refusal durably initializes the journal without reserving + a tag, so unrelated allocations can continue using the same journal. +- With headroom 1–12, an oversized preferred tag becomes one deterministic + hash-derived legal tag, no longer than headroom. Collisions return + `TAG_RESERVED` or `TAG_OCCUPIED`; there is no retry search or overwrite. +- Headroom at or below zero returns `Refusal('TAG_IMPOSSIBLE')`. Invalid input + and unreadable, corrupt, incompatible or replaced state also return explicit + refusals. Callers must branch on the result type and stop on every refusal; + truthiness is not a success check. + +Use an absent journal path in an existing private directory for the first +allocation. Only this module creates the mode-0600 file. Every process in the +same job must share it; preserve it across restarts. Never unlink, replace, +pre-create, truncate, copy to Git or reset it to recover from a failure. Missing +state after a restart requires operator reconciliation, not a fresh allocation. +The allocator holds a Linux `flock`, reads the current journal after locking, +checks the locked file's identity, then flushes/fsyncs the reservation before +returning. An interrupted in-place write can invalidate the journal and must +stop subsequent allocation. The private directory and its parent must remain +under exclusive caller control; this is not a cross-host lock or cloud lease. + +The journal binds project/region/service, operation, candidate and preferred tag +using full hashes, retaining only the operational tag as plaintext. Hashes are +private correlation data, not anonymisation. Returned tags are operational +values; do not log them. Dataclass representations suppress operational fields; +refusal messages contain static codes and never raw exceptions. For any caller +logs or API output, use the existing shared masking helper, including +`quant_platform_kit.notifications._redaction.mask_identifier` for resource +identifiers. Never mask an identifier before budgeting or identity comparisons. + +The caller's `existing_tags` maps **every** tag to its exact revision, including +zero-percent entries. An unknown revision is `None` and remains occupied. +Allocation leaves this mapping untouched. An allocation recovered after restart +is the same reservation, not an execution or replay permit. No release, expiry +or ownership-stealing operation is provided. + +## Impossible-tag lifecycle + +`lifecycle_for_impossible_tag` defaults to `STOP` and never claims an untagged +zero-percent candidate was warmed. A `MICRO_SPLIT` result proposes exactly +previous=99 / candidate=1, only when all these conditions hold: + +- The service has no nonempty tag headroom. +- The caller verified an authentic owner decision and supplied a + `WarmupApproval` with `warmup_traffic_approved=True` (a literal boolean). +- Project, region, service, action=`warmup`, and operation/run identity match + exactly. An approval for promotion, another run or another service cannot + widen the scope. +- `no_traffic` is explicitly `False` and `image_only` remains `False`; truthy + strings and unknown values cannot satisfy either gate. + +An approval object is policy input, not authentication. Callers must verify its +owner, validity, exact candidate and scope at their existing protected entry +point. Strict no-traffic diagnostics and image-only callers always stop on +impossible tags; this helper cannot grant them micro-traffic authority. + +## Caller mutation and restore contract + +Before any cloud write, the caller must hold a current exclusive execution +claim covering the exact target/run/candidate, durably record a single-use +mutation intent, and re-read the current service generation, image/source, +revision identity, traffic distribution and complete tags. The local allocation +journal does not serialize cloud writes across different jobs or hosts. Bind +approval to the exact candidate; do not reuse an operation ID for another image +or revision. Authenticate evidence using the caller's existing gates. + +Preserve all pre-existing tags and traffic entries, including zero-percent tags, +and compare the current cloud state against the approved before-image before +every stage. Use the caller's existing conditional-write/ownership mechanism; +a preflight read alone cannot prevent cloud races. A lost claim, state drift, +unknown write result or timeout requires readback and reconciliation, never an +automatic replay. A matching allocated tag is not permission to overwrite it. + +For an explicitly approved micro-split, verify exactly the permitted 99/1 split +and candidate startup/health. Restore the approved previous distribution on +failure, preserving every existing tag, then read it back before cleanup or a +subsequent stage. If ownership is lost or restoration cannot be verified, stop +all further writes and report `RESTORE_UNVERIFIED`. The policy supplies neither +gcloud mutation commands nor a claim that restore is atomic or always possible. + +## Adoption + +IBKR, Firstrade, LongBridge and QRS adopters must pin the reviewed QPK revision +or vendor this module with its contract version and offline tests. Replace +local formulas only within separately authorized caller changes, retaining the +entry point's existing traffic restrictions and identity gates. Confirm actual +caller use and pinned version before claiming adoption. This module alone +changes no deployed service or runtime behavior. Long-name service migration +remains deferred. diff --git a/src/quant_platform_kit/operators/__init__.py b/src/quant_platform_kit/operators/__init__.py new file mode 100644 index 00000000..1631402f --- /dev/null +++ b/src/quant_platform_kit/operators/__init__.py @@ -0,0 +1,17 @@ +"""Pinned operator policies; no cloud mutations or runtime trading imports.""" +from .cloud_run_tags import ( + CONTRACT_VERSION, + AllocatedTag, + Lifecycle, + Refusal, + Target, + WarmupApproval, + allocate_tag, + headroom, + lifecycle_for_impossible_tag, +) + +__all__ = [ + 'CONTRACT_VERSION', 'AllocatedTag', 'Lifecycle', 'Refusal', 'Target', + 'WarmupApproval', 'allocate_tag', 'headroom', 'lifecycle_for_impossible_tag', +] diff --git a/src/quant_platform_kit/operators/cloud_run_tags.py b/src/quant_platform_kit/operators/cloud_run_tags.py new file mode 100644 index 00000000..1056a9b0 --- /dev/null +++ b/src/quant_platform_kit/operators/cloud_run_tags.py @@ -0,0 +1,230 @@ +"""Versioned offline Cloud Run tag policy. This module never calls a cloud API. + +Allocation is an exclusive local reservation, NOT permission to mutate a service. +See docs/cloud-run-tags.md for inventory, ownership and restore requirements. +""" +from __future__ import annotations + +import fcntl +import hashlib +import json +import os +import re +import stat +from collections.abc import Mapping +from dataclasses import dataclass, field +from pathlib import Path + +CONTRACT_VERSION = 'cloud-run-tags/v1' +TAGGED_URL_BUDGET = 46 +_NAME = re.compile(r'[a-z](?:[a-z0-9-]*[a-z0-9])?\Z') +_DIGEST = re.compile(r'[0-9a-f]{64}\Z') + + +@dataclass(frozen=True) +class Refusal: + """Explicit failure with a static, identifier-free reason code.""" + reason: str + + +@dataclass(frozen=True) +class AllocatedTag: + """Operational value: do not log the tag or serialize it into public output.""" + tag: str = field(repr=False) + + +@dataclass(frozen=True) +class Target: + project: str = field(repr=False) + region: str = field(repr=False) + service_name: str = field(repr=False) + + +@dataclass(frozen=True) +class WarmupApproval: + """Caller-verified owner decision; this value cannot authenticate an approval.""" + target: Target = field(repr=False) + action: str = field(repr=False) + operation_id: str = field(repr=False) + warmup_traffic_approved: bool = field(repr=False) + + +@dataclass(frozen=True) +class Lifecycle: + kind: str + reason: str + previous_percent: int = 0 + candidate_percent: int = 0 + + +def headroom(service_name: str) -> int: + """Measure the exact original name, before any display masking.""" + return TAGGED_URL_BUDGET - len(service_name) + + +def _nonempty(value: object) -> bool: + return isinstance(value, str) and bool(value.strip()) and value == value.strip() + + +def _name(value: object) -> bool: + return isinstance(value, str) and bool(_NAME.fullmatch(value)) + + +def _target_valid(target: object) -> bool: + return (isinstance(target, Target) and _nonempty(target.project) + and _nonempty(target.region) and _name(target.service_name)) + + +def lifecycle_for_impossible_tag( + target: Target, action: str, operation_id: str, *, + approval: WarmupApproval | None = None, no_traffic: bool = True, + image_only: bool = False, +) -> Lifecycle: + """Default STOP; permit only an exact, explicitly approved warmup 99/1 plan. + + A returned plan still requires the caller's exclusive, single-use execution + claim and fresh cloud readback. It never grants deployment/traffic authority. + """ + if not _target_valid(target) or not _nonempty(operation_id): + return Lifecycle('STOP', 'INVALID_INPUT') + if no_traffic is not False or image_only is not False: + return Lifecycle('STOP', 'NO_TRAFFIC_REQUIRED') + if headroom(target.service_name) > 0: + return Lifecycle('STOP', 'TAG_NOT_IMPOSSIBLE') + if (action != 'warmup' or not isinstance(approval, WarmupApproval) + or approval.warmup_traffic_approved is not True + or approval.target != target or approval.action != action + or approval.operation_id != operation_id): + return Lifecycle('STOP', 'EXACT_WARMUP_APPROVAL_REQUIRED') + return Lifecycle('MICRO_SPLIT', 'EXACT_WARMUP_APPROVAL', 99, 1) + + +def _digest(*parts: str) -> str: + return hashlib.sha256(json.dumps(parts, separators=(',', ':')).encode()).hexdigest() + + +def _short_tag(scope: str, operation: str, candidate: str, size: int) -> str: + digest = _digest(CONTRACT_VERSION, scope, operation, candidate) + # Always begin with a letter, including the one-character boundary. + return (chr(ord('a') + int(digest[:2], 16) % 26) + digest[2:])[:size] + + +def _journal_valid(data: object) -> bool: + if (not isinstance(data, dict) or set(data) != {'contract_version', 'scopes'} + or data['contract_version'] != CONTRACT_VERSION + or not isinstance(data['scopes'], dict)): + return False + for scope, records in data['scopes'].items(): + if (not isinstance(scope, str) or not _DIGEST.fullmatch(scope) + or not isinstance(records, dict)): + return False + tags = set() + for operation, record in records.items(): + if (not isinstance(operation, str) or not _DIGEST.fullmatch(operation) + or not isinstance(record, dict) + or set(record) != {'candidate', 'preferred', 'tag'} + or any(not isinstance(record[k], str) or not _DIGEST.fullmatch(record[k]) + for k in ('candidate', 'preferred')) + or not _name(record['tag']) or record['tag'] in tags): + return False + tags.add(record['tag']) + return True + + +def _same_file(path: Path, fd: int) -> bool: + current = os.stat(path, follow_symlinks=False) + held = os.fstat(fd) + return (stat.S_ISREG(current.st_mode) and current.st_nlink == 1 + and (current.st_dev, current.st_ino) == (held.st_dev, held.st_ino) + and stat.S_IMODE(current.st_mode) == 0o600) + + +def _reserve(data: dict, scope: str, operation: str, candidate: str, + preferred: str, tag: str, existing: Mapping) -> AllocatedTag | Refusal: + records = data['scopes'].setdefault(scope, {}) + record = {'candidate': candidate, 'preferred': preferred, 'tag': tag} + previous = records.get(operation) + if previous is not None and previous != record: + return Refusal('OPERATION_REBOUND') + if any(key != operation and value['tag'] == tag for key, value in records.items()): + return Refusal('TAG_RESERVED') + # A cloud revision match alone does not prove operation ownership. + if tag in existing and (previous != record or _digest(existing[tag]) != candidate): + return Refusal('TAG_OCCUPIED') + records[operation] = record + return AllocatedTag(tag) + + +def allocate_tag( + service_name: str, preferred_tag: str, *, existing_tags: Mapping[str, str | None], + operation_id: str, candidate_id: str, state_path: str | Path, + project: str, region: str, +) -> AllocatedTag | Refusal: + """Reserve preferred or one deterministic shortened tag; never overwrite. + + existing_tags must come from a fresh, complete inventory (including zero- + percent entries); None as a value is occupied with unknown ownership. Use + one private, shared job-local state_path for all processes/restarts. Only + the allocator creates this file. Missing evidence or state I/O fails closed. + """ + if (not _target_valid(Target(project, region, service_name)) + or not _name(preferred_tag) or not _nonempty(operation_id) + or not _nonempty(candidate_id) or not isinstance(existing_tags, Mapping) + or any(not _name(k) or (v is not None and not _nonempty(v)) + for k, v in existing_tags.items())): + return Refusal('INVALID_INPUT') + room = headroom(service_name) + if room <= 0: + return Refusal('TAG_IMPOSSIBLE') + scope = _digest(project, region, service_name) + operation, candidate, preferred = map(_digest, (operation_id, candidate_id, preferred_tag)) + tag = preferred_tag if len(preferred_tag) <= room else _short_tag(scope, operation, candidate, room) + # Snapshot caller input without ever mutating it. Cloud freshness remains a + # caller obligation; the reservation ledger is read only AFTER flock. + existing = dict(existing_tags) + fd = None + try: + path = Path(state_path) + try: + fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + created = True + except FileExistsError: + fd = os.open(path, os.O_RDWR | os.O_NOFOLLOW) + created = False + with os.fdopen(fd, 'r+', encoding='utf-8') as stream: + fd = None + fcntl.flock(stream.fileno(), fcntl.LOCK_EX) + if not _same_file(path, stream.fileno()): + return Refusal('STATE_UNAVAILABLE') + text = stream.read() + # Another process may have initialized a just-created file first. + data = json.loads(text) if text else ({'contract_version': CONTRACT_VERSION, 'scopes': {}} if created else None) + if not _journal_valid(data): + return Refusal('STATE_UNAVAILABLE') + result = _reserve(data, scope, operation, candidate, preferred, tag, existing) + # Persist a newly initialized journal even on occupancy refusal. + # An empty existing journal still fails validation above. + if isinstance(result, Refusal) and not (created and not text): + return result + if not _same_file(path, stream.fileno()): + return Refusal('STATE_UNAVAILABLE') + stream.seek(0) + json.dump(data, stream, sort_keys=True, separators=(',', ':')) + stream.truncate() + stream.flush() + os.fsync(stream.fileno()) + # Persist the directory entry before returning a usable allocation. + directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + if not _same_file(path, stream.fileno()): + return Refusal('STATE_UNAVAILABLE') + return result + except (OSError, ValueError, TypeError, RecursionError): + # Do not include exception text: paths/resources may identify accounts. + return Refusal('STATE_UNAVAILABLE') + finally: + if fd is not None: + os.close(fd) diff --git a/tests/test_cloud_run_tags.py b/tests/test_cloud_run_tags.py new file mode 100644 index 00000000..4686c3a4 --- /dev/null +++ b/tests/test_cloud_run_tags.py @@ -0,0 +1,274 @@ +"""Offline contract tests using fake resource and operation identities.""" +from __future__ import annotations + +import importlib +import json +import multiprocessing +from pathlib import Path + +import pytest + + +def api(): + return importlib.import_module('quant_platform_kit.operators.cloud_run_tags') + + +def allocate(tmp_path, **overrides): + options = dict( + service_name='example-service', preferred_tag='cand-abcdef01', + existing_tags={}, operation_id='fake-run-one', candidate_id='fake-revision-one', + project='', region='example-region', state_path=tmp_path / 'tags.json', + ) + options.update(overrides) + return api().allocate_tag(**options) + + +def test_preferred_tag_and_exact_headroom(tmp_path): + result = allocate(tmp_path) + assert isinstance(result, api().AllocatedTag) + assert result.tag == 'cand-abcdef01' + assert api().headroom('example-service') == 31 + assert api().headroom('x' * 33) == 13 + + +@pytest.mark.parametrize('room', range(1, 13)) +def test_shorten_every_tight_budget_and_recover_after_restart(tmp_path, room): + result = allocate(tmp_path, service_name='x' * (46 - room)) + assert isinstance(result, api().AllocatedTag) + assert 1 <= len(result.tag) <= room + assert result.tag[0].isalpha() and result.tag.isalnum() + assert allocate(tmp_path, service_name='x' * (46 - room)) == result + assert json.loads((tmp_path / 'tags.json').read_text())['contract_version'] == api().CONTRACT_VERSION + + +@pytest.mark.parametrize('length', [46, 47, 48, 49, 63]) +def test_impossible_never_returns_silent_success(tmp_path, length): + result = allocate(tmp_path, service_name='x' * length) + assert result == api().Refusal('TAG_IMPOSSIBLE') + assert not (tmp_path / 'tags.json').exists() + + +@pytest.mark.parametrize('occupant', ['foreign-revision', None]) +def test_foreign_or_unknown_preferred_tag_refused(tmp_path, occupant): + assert allocate(tmp_path, existing_tags={'cand-abcdef01': occupant}) == api().Refusal('TAG_OCCUPIED') + + +def test_cloud_revision_match_without_local_ownership_is_not_owned(tmp_path): + assert allocate(tmp_path, existing_tags={'cand-abcdef01': 'fake-revision-one'}) == api().Refusal('TAG_OCCUPIED') + + +@pytest.mark.parametrize('occupant', [None, 'fake-foreign-revision', 'fake-revision-one']) +def test_initial_occupancy_refusal_keeps_journal_usable(tmp_path, occupant): + tags = {'cand-abcdef01': occupant} + assert allocate(tmp_path, existing_tags=tags) == api().Refusal('TAG_OCCUPIED') + data = json.loads((tmp_path / 'tags.json').read_text()) + assert api()._journal_valid(data) + assert not any(data['scopes'].values()) + assert tags == {'cand-abcdef01': occupant} + assert isinstance(allocate(tmp_path, service_name='unrelated-service'), api().AllocatedTag) + assert isinstance(allocate(tmp_path, preferred_tag='other-tag', operation_id='fake-run-two'), api().AllocatedTag) + + +def test_initial_occupancy_refusal_requires_durable_journal(tmp_path, monkeypatch): + def fail_fsync(_): + raise OSError('fake-private-path') + + monkeypatch.setattr(api().os, 'fsync', fail_fsync) + assert allocate(tmp_path, existing_tags={'cand-abcdef01': None}) == api().Refusal('STATE_UNAVAILABLE') + + +def test_owned_tag_and_zero_percent_tags_are_preserved(tmp_path): + allocated = allocate(tmp_path) + # Tag inventories include ALL entries, including zero-percent tags. + tags = {allocated.tag: 'fake-revision-one', 'zero-percent': 'fake-previous'} + before = dict(tags) + assert allocate(tmp_path, existing_tags=tags) == allocated + assert tags == before + assert allocate(tmp_path, existing_tags={allocated.tag: 'foreign-revision'}) == api().Refusal('TAG_OCCUPIED') + + +def test_hash_collision_refused_without_searching_another_tag(tmp_path): + first = allocate(tmp_path, service_name='x' * 45) + assert allocate(tmp_path, service_name='x' * 45, existing_tags={first.tag: None}) == api().Refusal('TAG_OCCUPIED') + + +def test_operation_identity_cannot_be_rebound_after_restart(tmp_path): + allocate(tmp_path) + assert allocate(tmp_path, candidate_id='fake-revision-two') == api().Refusal('OPERATION_REBOUND') + assert allocate(tmp_path, preferred_tag='other') == api().Refusal('OPERATION_REBOUND') + + +def test_latest_disk_reservation_wins_over_free_cloud_snapshot(tmp_path): + allocate(tmp_path) + assert allocate(tmp_path, operation_id='fake-run-two') == api().Refusal('TAG_RESERVED') + + +def test_missing_or_corrupt_inventory_and_journal_fail_closed(tmp_path): + assert allocate(tmp_path, existing_tags=None) == api().Refusal('INVALID_INPUT') + path = tmp_path / 'tags.json' + path.write_text('{incomplete') + assert allocate(tmp_path) == api().Refusal('STATE_UNAVAILABLE') + path.write_text('{}') + assert allocate(tmp_path) == api().Refusal('STATE_UNAVAILABLE') + + +def test_truncated_existing_journal_is_not_reinitialized(tmp_path): + allocate(tmp_path) + (tmp_path / 'tags.json').write_text('') + assert allocate(tmp_path) == api().Refusal('STATE_UNAVAILABLE') + + +def test_state_does_not_disclose_resource_or_candidate_identity(tmp_path): + result = allocate(tmp_path, service_name='fake-sensitive-service', candidate_id='fake-sensitive-candidate', operation_id='fake-sensitive-run') + data = (tmp_path / 'tags.json').read_text() + for value in ['fake-sensitive-service', 'fake-sensitive-candidate', 'fake-sensitive-run', '', 'example-region']: + assert value not in data + assert value not in repr(result) + target = api().Target('', 'example-region', 'fake-sensitive-service') + assert 'fake-sensitive-service' not in repr(target) + assert 'fake-sensitive-run' not in repr(api().WarmupApproval(target, 'warmup', 'fake-sensitive-run', True)) + + +def _allocate_process(path, run, queue): + result = allocate(Path(path), operation_id=run) + queue.put(type(result).__name__) + + +def test_two_processes_cannot_reserve_same_free_tag(tmp_path): + ctx = multiprocessing.get_context('spawn') + queue = ctx.Queue() + workers = [ctx.Process(target=_allocate_process, args=(str(tmp_path), f'fake-run-{n}', queue)) for n in range(2)] + for worker in workers: + worker.start() + for worker in workers: + worker.join(15) + assert worker.exitcode == 0 + assert sorted(queue.get(timeout=2) for _ in workers) == ['AllocatedTag', 'Refusal'] + + +def test_symlink_state_is_refused(tmp_path): + other = tmp_path / 'other.json' + other.write_text('') + (tmp_path / 'tags.json').symlink_to(other) + assert allocate(tmp_path) == api().Refusal('STATE_UNAVAILABLE') + assert other.read_text() == '' + + +@pytest.mark.parametrize('changes', [ + {'service_name': ' uppercase'}, {'preferred_tag': ''}, {'preferred_tag': '9bad'}, + {'operation_id': ''}, {'candidate_id': ''}, {'region': ''}, {'project': ''}, + {'existing_tags': {'bad tag': None}}, {'existing_tags': {'valid': 7}}, +]) +def test_invalid_inputs_refused(tmp_path, changes): + assert allocate(tmp_path, **changes) == api().Refusal('INVALID_INPUT') + + +def lifecycle(**overrides): + module = api() + target = module.Target('', 'example-region', 'x' * 48) + options = dict(target=target, action='warmup', operation_id='fake-run', no_traffic=False) + options.update(overrides) + return module.lifecycle_for_impossible_tag(**options) + + +def approval(**overrides): + module = api() + options = dict(target=module.Target('', 'example-region', 'x' * 48), action='warmup', operation_id='fake-run', warmup_traffic_approved=True) + options.update(overrides) + return module.WarmupApproval(**options) + + +def test_impossible_lifecycle_stops_by_default(): + assert lifecycle().kind == 'STOP' + assert lifecycle().candidate_percent == 0 + + +def test_only_exact_explicit_approval_selects_micro_split(): + result = lifecycle(approval=approval()) + assert result.kind == 'MICRO_SPLIT' + assert (result.previous_percent, result.candidate_percent) == (99, 1) + + +@pytest.mark.parametrize('changes', [ + {'target': lambda m: m.Target('', 'example-region', 'different-service')}, + {'target': lambda m: m.Target('', 'different-region', 'x' * 48)}, + {'target': lambda m: m.Target('fake-other-project', 'example-region', 'x' * 48)}, + {'action': lambda m: 'promotion'}, {'operation_id': lambda m: 'another-run'}, + {'warmup_traffic_approved': lambda m: False}, + {'warmup_traffic_approved': lambda m: 'true'}, +]) +def test_approval_does_not_widen(changes): + assert lifecycle(approval=approval(**{key: value(api()) for key, value in changes.items()})).kind == 'STOP' + + +@pytest.mark.parametrize('changes', [{'no_traffic': True}, {'image_only': True}, {'no_traffic': None}, {'image_only': None}]) +def test_no_traffic_and_image_only_never_select_micro_split(changes): + result = lifecycle(approval=approval(), **changes) + assert result.kind == 'STOP' + assert result.candidate_percent == 0 + + +def test_fitting_service_and_other_action_cannot_use_fallback(): + target = api().Target('', 'example-region', 'short-service') + assert lifecycle(target=target, approval=approval(target=target)).kind == 'STOP' + assert lifecycle(action='promotion', approval=approval(action='promotion')).kind == 'STOP' + + +def test_unknown_journal_version_is_refused(tmp_path): + allocate(tmp_path) + path = tmp_path / 'tags.json' + data = json.loads(path.read_text()) + data['contract_version'] = 'unknown' + path.write_text(json.dumps(data)) + assert allocate(tmp_path) == api().Refusal('STATE_UNAVAILABLE') + + +def test_replaced_lock_identity_blocks_reservation(tmp_path, monkeypatch): + module = api() + original = module._reserve + + def replace_state(*args): + result = original(*args) + path = tmp_path / 'tags.json' + path.rename(tmp_path / 'lost-lock.json') + path.write_text('replacement') + path.chmod(0o600) + return result + + monkeypatch.setattr(module, '_reserve', replace_state) + assert allocate(tmp_path) == module.Refusal('STATE_UNAVAILABLE') + assert (tmp_path / 'tags.json').read_text() == 'replacement' + + +def test_disk_error_does_not_expose_identifier(tmp_path, monkeypatch): + def fail_fsync(_): + raise OSError('fake-private-path') + + monkeypatch.setattr(api().os, 'fsync', fail_fsync) + result = allocate(tmp_path) + assert result == api().Refusal('STATE_UNAVAILABLE') + assert 'fake-private-path' not in repr(result) + + +def test_shortened_tags_also_keep_foreign_tags_intact(tmp_path): + result = allocate(tmp_path, service_name='x' * 40) + tags = {result.tag: 'fake-foreign-revision', 'zero-percent': None} + before = dict(tags) + assert allocate(tmp_path, service_name='x' * 40, existing_tags=tags) == api().Refusal('TAG_OCCUPIED') + assert tags == before + + +def test_different_operations_that_hash_to_one_character_collide(tmp_path): + module = api() + seen = {} + for n in range(27): + run = f'fake-run-{n}' + tag = module._short_tag(module._digest('', 'example-region', 'x' * 45), module._digest(run), module._digest('fake-revision-one'), 1) + if tag in seen: + first = allocate(tmp_path, service_name='x' * 45, operation_id=seen[tag]) + assert first.tag == tag + assert allocate(tmp_path, service_name='x' * 45, operation_id=run) == module.Refusal('TAG_RESERVED') + break + seen[tag] = run + else: + pytest.fail('27 operations must collide in a 26-letter namespace')