|
17 | 17 | """Tests for Broadcast Manager.""" |
18 | 18 |
|
19 | 19 |
|
| 20 | +import pytest |
| 21 | +from cylc.flow.cycling.integer import IntegerInterval, IntegerPoint |
| 22 | +from cylc.flow.cycling.iso8601 import ISO8601Interval, ISO8601Point |
| 23 | + |
| 24 | + |
20 | 25 | async def test_reject_valid_broadcast_is_remote_clash_with_config( |
21 | 26 | one_conf, flow, start, scheduler, log_filter |
22 | 27 | ): |
@@ -90,3 +95,120 @@ async def test_reject_valid_broadcast_is_remote_clash_with_broadcast( |
90 | 95 | {'platform': 'foo'}, |
91 | 96 | ] |
92 | 97 | } |
| 98 | + |
| 99 | + |
| 100 | +@pytest.mark.parametrize('cycling_mode', ('integer', 'gregorian', '360_day')) |
| 101 | +async def test_broadcast_expire_limit( |
| 102 | + cycling_mode, |
| 103 | + flow, |
| 104 | + scheduler, |
| 105 | + run, |
| 106 | + complete, |
| 107 | + capcall, |
| 108 | +): |
| 109 | + """Test automatic broadcast expiry. |
| 110 | +
|
| 111 | + To prevent broadcasts from piling up and causing a memory leak, we expire |
| 112 | + (aka clear) them. |
| 113 | +
|
| 114 | + The broadcast expiry limit is the oldest active cycle MINUS the longest |
| 115 | + cycling sequence. |
| 116 | +
|
| 117 | + See https://github.com/cylc/cylc-flow/pull/6964 |
| 118 | + """ |
| 119 | + # capture broadcast expiry calls |
| 120 | + _expires = capcall('cylc.flow.broadcast_mgr.BroadcastMgr.expire_broadcast') |
| 121 | + |
| 122 | + def expires(): |
| 123 | + """Return a list of the cycle limit expired since the last call.""" |
| 124 | + ret = [x[0][1] for x in _expires] |
| 125 | + _expires.clear() |
| 126 | + return ret |
| 127 | + |
| 128 | + def cycle(number): |
| 129 | + """Return a cycle point object in the relevant format.""" |
| 130 | + if cycling_mode == 'integer': |
| 131 | + return IntegerPoint(str(number)) |
| 132 | + else: |
| 133 | + return ISO8601Point(f'000{number}') |
| 134 | + |
| 135 | + def interval(number): |
| 136 | + """Return an integer object in the relevant format.""" |
| 137 | + if cycling_mode == 'integer': |
| 138 | + return IntegerInterval(sequence(number)) |
| 139 | + else: |
| 140 | + return ISO8601Interval(sequence(number)) |
| 141 | + |
| 142 | + def sequence(number): |
| 143 | + """Return a sequence string in the relevant format.""" |
| 144 | + if cycling_mode == 'integer': |
| 145 | + return f'P{number}' |
| 146 | + else: |
| 147 | + return f'P{number}Y' |
| 148 | + |
| 149 | + # a workflow with a sequential task |
| 150 | + id_ = flow({ |
| 151 | + 'scheduler': { |
| 152 | + 'cycle point format': 'CCYY' |
| 153 | + } if cycling_mode != 'integer' else {}, |
| 154 | + |
| 155 | + 'scheduling': { |
| 156 | + 'cycling mode': cycling_mode, |
| 157 | + 'initial cycle point': cycle(1), |
| 158 | + 'graph': { |
| 159 | + # the sequence with the sequential task |
| 160 | + sequence(1): f'a[-{sequence(1)}] => a', |
| 161 | + # a longer sequence to make the offset more interesting |
| 162 | + sequence(3): 'a', |
| 163 | + } |
| 164 | + } |
| 165 | + }) |
| 166 | + schd = scheduler(id_, paused_start=False) |
| 167 | + |
| 168 | + async with run(schd): |
| 169 | + # the longest cycling sequence has a step of "3" |
| 170 | + assert schd.config.interval_of_longest_sequence == interval(3) |
| 171 | + |
| 172 | + # no broadcast expires should happen on startup |
| 173 | + assert expires() == [] |
| 174 | + |
| 175 | + # when a cycle closes, auto broadcast expiry should happen |
| 176 | + # NOTE: datetimes cannot be negative, so this expiry will be skipped |
| 177 | + # for datetimetime cycling workflows |
| 178 | + await complete(schd, f'{cycle(1)}/a') |
| 179 | + assert expires() in ([], [cycle(-1)]) |
| 180 | + |
| 181 | + await complete(schd, f'{cycle(2)}/a') |
| 182 | + assert expires() == [cycle(0)] |
| 183 | + |
| 184 | + await complete(schd, f'{cycle(3)}/a') |
| 185 | + assert expires() == [cycle(1)] |
| 186 | + |
| 187 | + |
| 188 | +async def test_broadcast_expiry_async( |
| 189 | + one_conf, flow, scheduler, run, complete, capcall |
| 190 | +): |
| 191 | + """Test auto broadcast expiry with async workflows. |
| 192 | +
|
| 193 | + Auto broadcast expiry should not happen in async workflows as there is only |
| 194 | + one cycle so it doesn't make sense. |
| 195 | +
|
| 196 | + See https://github.com/cylc/cylc-flow/pull/6964 |
| 197 | + """ |
| 198 | + # capture broadcast expiry calls |
| 199 | + expires = capcall('cylc.flow.broadcast_mgr.BroadcastMgr.expire_broadcast') |
| 200 | + |
| 201 | + id_ = flow(one_conf) |
| 202 | + schd = scheduler(id_, paused_start=False) |
| 203 | + |
| 204 | + async with run(schd): |
| 205 | + # this is an async workflow so the longest cycling interval is a |
| 206 | + # null interval |
| 207 | + assert ( |
| 208 | + schd.config.interval_of_longest_sequence |
| 209 | + == IntegerInterval.get_null() |
| 210 | + ) |
| 211 | + await complete(schd) |
| 212 | + |
| 213 | + # no auto-expiry should take place |
| 214 | + assert expires == [] |
0 commit comments