1dc1d2f3a1
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
114 lines
7.4 KiB
Python
114 lines
7.4 KiB
Python
import copy
|
|
from datetime import datetime, timedelta, timezone
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import pytest
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1] / 'scripts'
|
|
if not SCRIPTS.exists(): SCRIPTS = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
import workload_adoption_proof as proof
|
|
|
|
NOW = datetime(2026, 10, 11, 20, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def observation():
|
|
c = proof.CONTRACT
|
|
old = (NOW - timedelta(days=2)).isoformat()
|
|
settings = {'method': 'annotation', 'controller_version': 'v3.5.3', 'installation_id_absent': True}
|
|
return {'captured_at': NOW.isoformat(), 'metadata_captured_at': NOW.isoformat(), 'estate_ip': c['estate_ip'], 'tracking_settings': settings,
|
|
'application': {'name': c['application'], 'uid': 'application-uid', 'project': c['project'], 'destination': c['destination'],
|
|
'source': {'repoURL': c['repo_url'], 'path': c['source_path'], 'targetRevision': c['revision']}, 'revision': c['revision'],
|
|
'automated_present': False, 'pending_operation': False, 'sync': 'Synced', 'health': 'Healthy', 'condition_types': [],
|
|
'operation': {'phase': 'Succeeded', 'startedAt': old, 'finishedAt': old}, 'sync_prune': False,
|
|
'sync_strategy_apply': True, 'sync_strategy_force': False, 'operation_revision': c['revision'], 'resources': [{**r, 'status': 'Synced'} for r in c['resources']]},
|
|
'ready_checks': [{**check, 'uid': 'resource-uid', 'tracking_matches': True,
|
|
'ready': [{'status': 'True', 'lastTransitionTime': old}], 'generation': 2, 'observed_generation': 2,
|
|
'replicas': 1, 'available_replicas': 1, 'instances': 1, 'ready_instances': 1} for check in c['ready_checks']],
|
|
'live_hooks': [], 'public_health': {'url': c['health_url'], 'status': 200},
|
|
'pinned_render_diff': {'revision': c['revision'], 'path': c['source_path'], 'client_side_exit_code': 0,
|
|
'render_resource_count': len(c['resources']), 'skipped_hooks': sorted(c['hooks']), 'output_suppressed': True,
|
|
'render_sha256': 'a' * 64, 'normalized_render_sha256': 'b' * 64}}
|
|
|
|
|
|
def test_verified_receipt_is_manual_adoption_only():
|
|
receipt = proof.verify(observation(), proof.CONTRACT, now=NOW)
|
|
assert receipt['status'] == 'verified'
|
|
assert receipt['resource_count'] == 5
|
|
assert receipt['ready_count'] == 3
|
|
assert receipt['proving_seconds'] == 172800
|
|
assert receipt['native_mutation'] is False
|
|
assert receipt['child_automation_enabled'] is False
|
|
|
|
|
|
@pytest.mark.parametrize('field,value', [('automated_present', True), ('pending_operation', True), ('sync_prune', True),
|
|
('sync_strategy_apply', False), ('sync_strategy_force', True), ('revision', 'f'*40), ('health', 'Progressing'), ('condition_types', ['ComparisonError'])])
|
|
def test_wrong_runtime_gate_fails(field, value):
|
|
obj = observation(); obj['application'][field] = value
|
|
with pytest.raises(ValueError): proof.verify(obj, proof.CONTRACT, now=NOW)
|
|
|
|
|
|
@pytest.mark.parametrize('mutation', ['missing-resource', 'extra-hook-resource', 'resource-not-synced', 'hook-live',
|
|
'short-proving', 'stale-observation', 'future-observation', 'metadata-stale', 'client-diff', 'missing-check', 'ready-false',
|
|
'new-ready-transition', 'deployment-generation', 'deployment-replicas', 'cnpg-instances', 'tracking', 'health-url', 'health-status'])
|
|
def test_missing_or_wrong_proof_fails(mutation):
|
|
obj = observation()
|
|
if mutation == 'missing-resource': obj['application']['resources'].pop()
|
|
elif mutation == 'extra-hook-resource': obj['application']['resources'].append({'group': 'batch', 'kind': 'Job', 'namespace': 'target-revenue', 'name': proof.CONTRACT['hooks'][0], 'status': 'Synced'})
|
|
elif mutation == 'resource-not-synced': obj['application']['resources'][0]['status'] = 'OutOfSync'
|
|
elif mutation == 'hook-live': obj['live_hooks'] = [proof.CONTRACT['hooks'][0]]
|
|
elif mutation == 'short-proving': obj['application']['operation']['finishedAt'] = (NOW-timedelta(hours=23)).isoformat()
|
|
elif mutation == 'stale-observation': obj['captured_at'] = (NOW-timedelta(minutes=6)).isoformat()
|
|
elif mutation == 'future-observation': obj['captured_at'] = (NOW+timedelta(seconds=1)).isoformat()
|
|
elif mutation == 'metadata-stale': obj['metadata_captured_at'] = (NOW-timedelta(minutes=6)).isoformat()
|
|
elif mutation == 'client-diff': obj['pinned_render_diff']['client_side_exit_code'] = 1
|
|
elif mutation == 'missing-check': obj['ready_checks'].pop()
|
|
elif mutation == 'ready-false': obj['ready_checks'][1]['ready'][0]['status'] = 'False'
|
|
elif mutation == 'new-ready-transition': obj['ready_checks'][1]['ready'][0]['lastTransitionTime'] = NOW.isoformat()
|
|
elif mutation == 'deployment-generation': obj['ready_checks'][0]['observed_generation'] = 1
|
|
elif mutation == 'deployment-replicas': obj['ready_checks'][0]['available_replicas'] = 0
|
|
elif mutation == 'cnpg-instances': obj['ready_checks'][2]['ready_instances'] = 0
|
|
elif mutation == 'tracking': obj['ready_checks'][0]['tracking_matches'] = False
|
|
elif mutation == 'health-url': obj['public_health']['url'] = 'https://unexpected.test/healthz'
|
|
elif mutation == 'health-status': obj['public_health']['status'] = 503
|
|
with pytest.raises(ValueError): proof.verify(obj, proof.CONTRACT, now=NOW)
|
|
|
|
|
|
def sources():
|
|
docs = []
|
|
for r in proof.CONTRACT['resources']:
|
|
docs.append({'apiVersion': r['group']+'/v1' if r['group'] else 'v1', 'kind': r['kind'],
|
|
'metadata': {'name': r['name'], 'namespace': r['namespace']}})
|
|
for name in proof.CONTRACT['hooks']:
|
|
docs.append({'apiVersion': 'batch/v1', 'kind': 'Job', 'metadata': {'name': name, 'namespace': 'target-revenue', 'annotations': {'argocd.argoproj.io/hook': 'Sync'}}})
|
|
return docs
|
|
|
|
|
|
def test_normalization_skips_hooks_and_only_injects_tracking_metadata():
|
|
docs = sources(); original = copy.deepcopy(docs)
|
|
settings = observation()['tracking_settings']
|
|
normalized = proof.normalize(docs, proof.CONTRACT, settings)
|
|
assert len(normalized) == 5
|
|
assert docs == original
|
|
assert not any(r['kind'] == 'Job' for r in normalized)
|
|
assert normalized[0]['metadata']['annotations']['argocd.argoproj.io/tracking-id'] == 'target-revenue:/Service:target-revenue/target-revenue'
|
|
|
|
|
|
@pytest.mark.parametrize('bad', ['secret', 'extra-resource', 'wrong-hook', 'missing-hook', 'duplicate-resource', 'conflicting-tracking'])
|
|
def test_normalization_fails_closed(bad):
|
|
docs = sources()
|
|
if bad == 'secret': docs.append({'apiVersion': 'v1', 'kind': 'Secret', 'metadata': {'name': 'secret', 'namespace': 'target-revenue'}})
|
|
elif bad == 'extra-resource': docs.append({'apiVersion': 'v1', 'kind': 'Service', 'metadata': {'name': 'extra', 'namespace': 'target-revenue'}})
|
|
elif bad == 'wrong-hook': docs[-1]['metadata']['annotations']['argocd.argoproj.io/hook'] = 'PreSync'
|
|
elif bad == 'missing-hook': docs.pop()
|
|
elif bad == 'duplicate-resource': docs.append(copy.deepcopy(docs[0]))
|
|
elif bad == 'conflicting-tracking': docs[0]['metadata']['annotations'] = {'argocd.argoproj.io/tracking-id': 'other'}
|
|
with pytest.raises(ValueError): proof.normalize(docs, proof.CONTRACT, observation()['tracking_settings'])
|
|
|
|
|
|
def test_kubectl_stream_and_list_decoding():
|
|
docs = sources()
|
|
assert proof.decode_objects(('\n'.join(json.dumps(r) for r in docs)).encode()) == docs
|
|
assert proof.decode_objects(json.dumps({'kind': 'List', 'items': docs}).encode()) == docs
|