MCPcopy Create free account
hub / github.com/MaterializeInc/materialize / run

Method run

misc/python/materialize/zippy/framework.py:226–253  ·  view source on GitHub ↗

Run the Zippy test.

(self, c: Composition)

Source from the content-addressed store, hash-verified

224 return actions
225
226 def run(self, c: Composition) -> None:
227 """Run the Zippy test."""
228 max_time = datetime.now() + self._max_execution_time
229 executed_count = len(self._actions)
230 for i, action in enumerate(self._actions):
231 print(action)
232 action.run(c, self._state)
233 if datetime.now() > max_time:
234 print(
235 f"--- Desired execution time of {self._max_execution_time} has been reached."
236 )
237 executed_count = i + 1
238 break
239
240 # Remove capability effects of actions that were never executed,
241 # so that finalization only validates objects that actually exist.
242 for action in self._actions[executed_count:]:
243 for cap in action.provides():
244 self._capabilities.remove_capability_instance(cap)
245
246 # Generate finalization actions now, after the main loop, so that
247 # capabilities reflect what was actually created during execution.
248 for action_or_factory in self._scenario.finalization():
249 self._final_actions.extend(self.generate_actions(action_or_factory))
250
251 for action in self._final_actions:
252 print(action)
253 action.run(c, self._state)
254
255 def _pick_action_or_factory(self) -> ActionOrFactory:
256 """Select the next Action to run in the Test"""

Callers 5

workflow_defaultFunction · 0.95
mainFunction · 0.45

Calls 6

generate_actionsMethod · 0.95
nowMethod · 0.45
providesMethod · 0.45
finalizationMethod · 0.45
extendMethod · 0.45

Tested by

no test coverage detected