railiance-platform/tests/test_workload_adoption_proof.py

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