informed-decision/tests/test_deployment_candidate.py

import copy
import json
import importlib.util
from pathlib import Path

import pytest

spec=importlib.util.spec_from_file_location('render_deployment',Path(__file__).parents[1]/'tools/render_deployment.py')
module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)


def inputs():
    return {'image':'forgejo.coulomb.social/coulomb/informed-decision@sha256:'+'a'*64,
        'policy':{'origin':'http://flex-auth-review.flex-auth.svc.cluster.local:8080','package':'review.example',
            'version':'v1','package_digest':'sha256:'+'b'*64,'pod_name':'flex-auth-review'},
        'keycape_egress_ips':['92.205.62.239'],'storage_class':'local-path'}


def test_candidate_does_not_create_custody_and_uses_one_private_writer():
    result=module.render(inputs());objects=result['items']
    assert not any(o['kind'] in {'Secret','Role','RoleBinding','ClusterRoleBinding','Ingress','Namespace'} for o in objects)
    deploy=next(o for o in objects if o['kind']=='Deployment')['spec']
    assert deploy['replicas']==1 and deploy['strategy']=={'type':'Recreate'}
    pod=deploy['template']['spec'];container=pod['containers'][0]
    assert pod['automountServiceAccountToken'] is False and pod['serviceAccountName']=='review'
    assert pod['securityContext']['runAsUser']==pod['securityContext']['fsGroup']==10001
    assert container['securityContext']['readOnlyRootFilesystem'] is True
    assert container['securityContext']['capabilities']['drop']==['ALL']
    caller=next(v for v in pod['volumes'] if v['name']=='caller')
    assert caller['projected']['sources']==[{'serviceAccountToken':{'path':'token','audience':'flex-auth','expirationSeconds':3600}}]
    # Healthy serving remains available for refusal/recovery; application itself
    # gates accept on audit /readyz, without a dependency-driven restart cycle.
    assert container['readinessProbe']['httpGet']['path']=='/healthz'
    assert container['livenessProbe']['httpGet']['path']=='/healthz'


def test_policies_and_service_selectors_do_not_include_placeholder_or_whole_namespaces():
    objects=module.render(inputs())['items']
    service=next(o for o in objects if o['kind']=='Service')
    assert service['spec']['selector']['app.kubernetes.io/component']=='review'
    policy=next(o for o in objects if o['kind']=='NetworkPolicy')['spec']
    for rule in policy['ingress']+policy['egress']:
        for peer in rule.get('from',rule.get('to',[])):
            if 'namespaceSelector' in peer:assert 'podSelector' in peer
    assert policy['egress'][-1]['to']==[{'ipBlock':{'cidr':'92.205.62.239/32'}}]
    assert policy['egress'][-1]['ports']==[{'protocol':'TCP','port':443}]
    assert policy['egress'][-2]['to'][0]['podSelector']['matchLabels']=={'app.kubernetes.io/name':'traefik'}
    assert policy['egress'][-2]['ports']==[{'protocol':'TCP','port':8443}]
    peers=[o for o in objects if o['kind']=='NetworkPolicy' and o['metadata']['namespace']!='informed-decision']
    assert {o['metadata']['namespace'] for o in peers}=={'approval-engine','flex-auth'}
    for obj in peers:
        caller=obj['spec']['ingress'][0]['from'][0]
        assert caller['namespaceSelector']['matchLabels']=={'kubernetes.io/metadata.name':'informed-decision'}
        assert caller['podSelector']['matchLabels']['app.kubernetes.io/component']=='review'


@pytest.mark.parametrize('fault',['tag','other_image','missing_pin','external_pdp','wildcard_pod','unknown_field',
                                 'broad_egress','loopback','empty_ips','empty_storage','empty_package'])
def test_incomplete_or_broad_deployment_inputs_refuse(fault):
    data=inputs()
    if fault=='tag':data['image']='forgejo.coulomb.social/coulomb/informed-decision:latest'
    if fault=='other_image':data['image']=data['image'].replace('informed-decision','other')
    if fault=='missing_pin':del data['policy']['package_digest']
    if fault=='external_pdp':data['policy']['origin']='https://elsewhere.test'
    if fault=='wildcard_pod':data['policy']['pod_name']='*'
    if fault=='unknown_field':data['token']='not-permitted-inline'
    if fault=='broad_egress':data['keycape_egress_ips']=['0.0.0.0/0']
    if fault=='loopback':data['keycape_egress_ips']=['127.0.0.1']
    if fault=='empty_ips':data['keycape_egress_ips']=[]
    if fault=='empty_storage':data['storage_class']=''
    if fault=='empty_package':data['policy']['package']=''
    with pytest.raises(ValueError):module.render(data)


def test_policy_revision_changes_immutable_configuration_name():
    first=inputs();second=copy.deepcopy(first);second['policy']['version']='v2'
    names=lambda data:[o['metadata']['name'] for o in module.render(data)['items'] if o['kind']=='ConfigMap']
    assert names(first)!=names(second)


def test_optional_workspace_requires_distinct_pins_and_exact_policy_network_peer():
    data = inputs()
    data['lifecycle_policy'] = {**data['policy'], 'package': 'informed-decision.lifecycle',
                              'origin': 'http://flex-auth-lifecycle.flex-auth.svc.cluster.local:8080',
                              'pod_name': 'flex-auth-lifecycle'}
    with pytest.raises(ValueError):
        module.render(data)
    data['lifecycle_delivery'] = True
    objects = module.render(data)['items']
    config = json.loads(next(row for row in objects if row['kind']=='ConfigMap')['data']['runtime.json'])
    assert config['lifecycle_policy']['package']=='informed-decision.lifecycle'
    assert config['lifecycle_delivery'] is True
    review_network = next(row for row in objects if row['kind']=='NetworkPolicy' and row['metadata']['namespace']=='informed-decision')
    assert any(peer.get('podSelector', {}).get('matchLabels', {}).get('app.kubernetes.io/name')=='flex-auth-lifecycle'
               for rule in review_network['spec']['egress'] for peer in rule.get('to', []))
    assert any(row['metadata']['name']=='informed-decision-lifecycle-ingress' for row in objects)
    data['lifecycle_policy']['package'] = data['policy']['package']
    with pytest.raises(ValueError):
        module.render(data)


def workload_inputs():
    data = inputs()
    data['lifecycle_policy'] = {**data['policy'], 'package': 'informed-decision.lifecycle', 'pod_name': 'flex-auth-lifecycle'}
    data['lifecycle_delivery'] = True
    data['intake_attributes'] = True
    data['intake_delivery'] = {'hub_origin': 'http://state-hub.state-hub.svc.cluster.local:8000'}
    data['workload_api'] = {'intake': True, 'lifecycle': True,
        'registries_configmap': 'informed-decision-workload-registry',
        'peers': [{'namespace': 'state-hub', 'app': 'state-hub'}],
        'api_egress': [{'ip': '10.43.0.1', 'port': 443}, {'ip': '92.205.62.239', 'port': 6443}]}
    return data


def test_private_workload_candidate_has_exact_rbac_peers_and_projection():
    objects = module.render(workload_inputs())['items']
    services = [o for o in objects if o['kind'] == 'Service']
    assert len(services) == 2
    assert services[0]['spec']['ports'][0]['port'] == 80
    assert services[1]['spec']['ports'][0]['port'] == 8081
    assert all(o['kind'] not in ('Secret', 'Ingress') for o in objects)
    role = next(o for o in objects if o['kind'] == 'ClusterRole')
    assert role['rules'] == [{'apiGroups':['authentication.k8s.io'],'resources':['tokenreviews'],'verbs':['create']}]
    binding = next(o for o in objects if o['kind'] == 'ClusterRoleBinding')
    assert binding['subjects'] == [{'kind':'ServiceAccount','name':'review','namespace':'informed-decision'}]
    policy = next(o for o in objects if o['kind']=='NetworkPolicy' and o['metadata']['name']=='informed-decision-review')
    rule = policy['spec']['ingress'][-1]
    assert rule['ports'] == [{'protocol':'TCP','port':8081}]
    assert rule['from'][0]['namespaceSelector']['matchLabels'] == {'kubernetes.io/metadata.name':'state-hub'}
    pod = next(o for o in objects if o['kind']=='Deployment')['spec']['template']['spec']
    registry = next(m for m in pod['containers'][0]['volumeMounts'] if m['name']=='workload-registry')
    assert registry['readOnly'] and 'subPath' not in registry
    requester = next(v for v in pod['volumes'] if v['name']=='sitting-requester')
    assert requester['secret']['secretName'] == 'informed-decision-sitting-requester'
    assert requester['secret']['defaultMode'] == 0o440
    config = json.loads(next(o for o in objects if o['kind']=='ConfigMap')['data']['runtime.json'])
    assert config['workload_api']['token_review']['origin'] == 'https://kubernetes.default.svc:443'


@pytest.mark.parametrize('fault', ['wildcard','missing_delivery','no_policy','broad_api','false_type','empty_peers'])
def test_private_workload_candidate_refuses_incomplete_or_broad_configuration(fault):
    data = workload_inputs()
    if fault == 'wildcard': data['workload_api']['peers'][0]['app']='*'
    if fault == 'missing_delivery': del data['intake_delivery']
    if fault == 'no_policy': del data['lifecycle_policy']
    if fault == 'broad_api': data['workload_api']['api_egress'][0]['ip']='10.0.0.0/8'
    if fault == 'false_type': data['workload_api']['intake']='true'
    if fault == 'empty_peers': data['workload_api']['peers']=[]
    with pytest.raises(ValueError): module.render(data)