activity-core/tests/test_release_health.py

import copy
from datetime import timedelta
from types import SimpleNamespace
from uuid import UUID

import pytest
from temporalio.client import (Schedule, ScheduleActionStartWorkflow, ScheduleCalendarSpec,
    SchedulePolicy, ScheduleRange, ScheduleSpec, ScheduleState, ScheduleOverlapPolicy)
from temporalio.converter import DataConverter
from temporalio.api.common.v1 import Payloads, WorkflowType
from temporalio.api.taskqueue.v1 import TaskQueue
from temporalio.api.workflow.v1 import NewWorkflowExecutionInfo

from activity_core.release_health import FRONTEND, SINKS, definition_matches, probe, schedule_matches


def rows():
    return [SimpleNamespace(id=UUID(identity),enabled=True,trigger_type='cron',
        trigger_config={'cron_expression':cron,'timezone':'Europe/Berlin','misfire_policy':'catchup_latest','catchup_window_seconds':86400},
        instructions_json=[{'id':f'frontend-patterns-{cadence}-report','model':'deterministic','report_sinks':[dict(SINKS[cadence])]}])
        for identity,(cadence,cron) in FRONTEND.items()]


def description(identity,calendar=False):
    cadence,cron=FRONTEND[identity]
    action=ScheduleActionStartWorkflow('RunActivityWorkflow',args=[identity,'scheduled',None],
        id=f'activity-{identity}:${{firstScheduledTime}}',task_queue='orchestrator-tq')
    if calendar:
        cal=ScheduleCalendarSpec(hour=[ScheduleRange({'daily':2,'weekly':3,'monthly':9}[cadence])],
             day_of_week=[ScheduleRange(2)] if cadence=='weekly' else [ScheduleRange(0,6)],
             day_of_month=[ScheduleRange(1)] if cadence=='monthly' else [ScheduleRange(1,31)])
        spec=ScheduleSpec(calendars=[cal],time_zone_name='Europe/Berlin')
    else:spec=ScheduleSpec(cron_expressions=[cron],time_zone_name='Europe/Berlin')
    schedule=Schedule(action=action,spec=spec,policy=SchedulePolicy(overlap=ScheduleOverlapPolicy.BUFFER_ONE,
        catchup_window=timedelta(days=1)),state=ScheduleState(paused=False))
    return SimpleNamespace(schedule=schedule)


def sources(items=None,descriptions=None):
    data=items if items is not None else rows()
    class Session:
        async def __aenter__(self):return self
        async def __aexit__(self,*_):pass
        async def scalars(self,statement):
            assert len(statement.compile().params['id_1'])==3
            return SimpleNamespace(all=lambda:data)
    descriptions=descriptions or {identity:description(identity,calendar=True) for identity in FRONTEND}
    class Temporal:
        data_converter=DataConverter.default
        def get_schedule_handle(self,sid):
            assert sid.startswith('activity-schedule-')
            async def describe():return descriptions[sid.removeprefix('activity-schedule-')]
            return SimpleNamespace(describe=describe)
    return Session,Temporal()


@pytest.mark.parametrize('calendar',[False,True])
async def test_real_invariants_accept_exact_cron_and_server_normalized_calendar(calendar):
    session,temporal=sources(descriptions={identity:description(identity,calendar) for identity in FRONTEND})
    result=await probe(session,temporal)
    assert result['healthy'] is True and result['definition_count']==result['schedule_count']==3
    assert set(result)=={'schema_version','observed_at','healthy','definition_count','schedule_count','definitions_and_report_sinks_match','schedules_match'}


async def test_actual_describe_payload_arguments_are_decoded_with_bound_client_converter():
    descriptions={identity:description(identity,True) for identity in FRONTEND}
    for identity,value in descriptions.items():
        encoded=await DataConverter.default.encode([identity,'scheduled',None])
        raw=NewWorkflowExecutionInfo(workflow_id=f'activity-{identity}:${{firstScheduledTime}}',
            workflow_type=WorkflowType(name='RunActivityWorkflow'),task_queue=TaskQueue(name='orchestrator-tq'),input=Payloads(payloads=encoded))
        value.schedule.action=ScheduleActionStartWorkflow('<unused>',raw_info=raw)
    session,temporal=sources(descriptions=descriptions)
    assert (await probe(session,temporal))['healthy'] is True


@pytest.mark.parametrize('field,value',[('enabled',False),('trigger_type','event'),
    ('instructions_json',[{'model':'llm','report_sinks':[SINKS['daily']]}])])
async def test_definition_behavior_drift_refused(field,value):
    data=rows();setattr(data[0],field,value)
    assert (await probe(*sources(items=data)))['healthy'] is False


@pytest.mark.parametrize('field,value',[('cron_expression','* * * * *'),('timezone','UTC'),
    ('misfire_policy','catchup_all'),('catchup_window_seconds',3600),('jitter_seconds',5)])
def test_trigger_scope_drift_refused(field,value):
    row=rows()[0];row.trigger_config[field]=value
    assert not definition_matches(row)


async def test_live_rows_without_explicit_catchup_window_use_policy_default():
    data=rows()
    for row in data:del row.trigger_config['catchup_window_seconds']
    assert (await probe(*sources(items=data)))['healthy'] is True


def test_each_cadence_requires_its_own_report_sink():
    for row in rows():
        cadence=FRONTEND[str(row.id)][0]
        other=next(c for c in SINKS if c!=cadence)
        row.instructions_json[0]['report_sinks']=[dict(SINKS[other])]
        assert not definition_matches(row)


@pytest.mark.parametrize('change',['pause','limit','extra_calendar','interval','skip','cron','queue','args','sink','policy','timezone'])
async def test_schedule_and_report_drift_fail_closed(change):
    data=rows();descriptions={identity:description(identity,True) for identity in FRONTEND}
    identity=next(iter(FRONTEND));s=descriptions[identity].schedule
    if change=='pause':s.state.paused=True
    elif change=='limit':s.state.limited_actions=True
    elif change=='extra_calendar':s.spec.calendars.append(ScheduleCalendarSpec())
    elif change=='interval':s.spec.intervals=[object()]
    elif change=='skip':s.spec.skip=[ScheduleCalendarSpec()]
    elif change=='cron':s.spec.calendars[0].hour=[ScheduleRange(5)]
    elif change=='queue':s.action.task_queue='privileged-other-queue'
    elif change=='args':s.action.args=['other-identity','scheduled',None]
    elif change=='sink':data[0].instructions_json[0]['report_sinks'][0]['topic_id']='foreign'
    elif change=='policy':s.policy.overlap=ScheduleOverlapPolicy.ALLOW_ALL
    elif change=='timezone':s.spec.time_zone_name='UTC'
    assert (await probe(*sources(items=data,descriptions=descriptions)))['healthy'] is False


async def test_missing_duplicate_rows_and_private_exceptions_are_not_healthy_or_leaked():
    for data in [[],rows()[:2],[rows()[0]]*3]:assert not (await probe(*sources(items=data)))['healthy']
    def unavailable():raise RuntimeError('PRIVATE-DATABASE-PASSWORD-CANARY')
    result=await probe(unavailable,sources()[1])
    assert not result['healthy'] and 'CANARY' not in str(result)


def test_release_health_endpoint_is_read_only_and_uses_bound_dependencies(monkeypatch):
    from fastapi.testclient import TestClient
    from activity_core import api
    session,temporal=sources()
    monkeypatch.setattr(api,'_session_factory',session)
    monkeypatch.setattr(api,'_temporal_client',temporal)
    client=TestClient(api.app)
    result=client.get('/release/health')
    assert result.status_code==200 and result.json()['healthy']
    assert client.post('/release/health',json={'operation':'trigger'}).status_code==405