Module materialize.checks.actions
Expand source code Browse git
# Copyright Materialize, Inc. and contributors. All rights reserved.
#
# Use of this software is governed by the Business Source License
# included in the LICENSE file at the root of this repository.
#
# As of the Change Date specified in that file, in accordance with
# the Business Source License, use of this software will be governed
# by the Apache License, Version 2.0.
import textwrap
import time
from inspect import getframeinfo, stack
from typing import TYPE_CHECKING, Any
from materialize.checks.executors import Executor
if TYPE_CHECKING:
from materialize.checks.scenarios import Scenario
class Action:
def __init__(self) -> None:
self.mz_service = None
self.phase = None
def execute(self, e: Executor) -> None:
assert False
def join(self, e: Executor) -> None:
print(f"Action {self} does not implement join()")
raise NotImplementedError
class Testdrive(Action):
# Instruct pytest this class does not contain actual tests
__test__ = False
def __init__(self, input: str, dedent: bool = True) -> None:
self.input = textwrap.dedent(input) if dedent else input
self.handle: Any | None = None
self.caller = getframeinfo(stack()[1][0])
def execute(self, e: Executor, mz_service: str | None = None) -> None:
"""Pass testdrive actions to be run by an Executor-specific implementation."""
self.handle = e.testdrive(self.input, self.caller, mz_service)
def join(self, e: Executor) -> None:
e.join(self.handle)
class Sleep(Action):
def __init__(self, interval: float) -> None:
self.interval = interval
def execute(self, e: Executor) -> None:
print(f"Sleeping for {self.interval} seconds")
time.sleep(self.interval)
def join(self, e: Executor) -> None:
pass
class Initialize(Action):
def __init__(self, scenario: "Scenario", mz_service: str | None = None) -> None:
self.checks = scenario.check_objects
self.mz_service = mz_service
def execute(self, e: Executor) -> None:
for check in self.checks:
print(f"Running initialize() from {check}")
check.start_initialize(e, self)
def join(self, e: Executor) -> None:
for check in self.checks:
check.join_initialize(e)
class Manipulate(Action):
def __init__(
self,
scenario: "Scenario",
phase: int | None = None,
mz_service: str | None = None,
) -> None:
assert phase is not None
self.phase = phase - 1
self.mz_service = mz_service
self.checks = scenario.check_objects
def execute(self, e: Executor) -> None:
assert self.phase is not None
for check in self.checks:
print(f"Running manipulate() from {check}")
check.start_manipulate(e, self)
def join(self, e: Executor) -> None:
assert self.phase is not None
for check in self.checks:
check.join_manipulate(e, self)
class Validate(Action):
def __init__(self, scenario: "Scenario", mz_service: str | None = None) -> None:
self.checks = scenario.check_objects
self.mz_service = mz_service
def execute(self, e: Executor) -> None:
for check in self.checks:
print(f"Running validate() from {check}")
check.start_validate(e, self)
def join(self, e: Executor) -> None:
for check in self.checks:
check.join_validate(e)
Classes
class Action
-
Expand source code Browse git
class Action: def __init__(self) -> None: self.mz_service = None self.phase = None def execute(self, e: Executor) -> None: assert False def join(self, e: Executor) -> None: print(f"Action {self} does not implement join()") raise NotImplementedError
Subclasses
- Initialize
- Manipulate
- Sleep
- Testdrive
- Validate
- Backup
- Restore
- ReplaceEnvironmentdStatefulSet
- SetupSshTunnels
- MzcomposeAction
Methods
def execute(self, e: Executor) ‑> None
-
Expand source code Browse git
def execute(self, e: Executor) -> None: assert False
def join(self, e: Executor) ‑> None
-
Expand source code Browse git
def join(self, e: Executor) -> None: print(f"Action {self} does not implement join()") raise NotImplementedError
class Initialize (scenario: Scenario, mz_service: str | None = None)
-
Expand source code Browse git
class Initialize(Action): def __init__(self, scenario: "Scenario", mz_service: str | None = None) -> None: self.checks = scenario.check_objects self.mz_service = mz_service def execute(self, e: Executor) -> None: for check in self.checks: print(f"Running initialize() from {check}") check.start_initialize(e, self) def join(self, e: Executor) -> None: for check in self.checks: check.join_initialize(e)
Ancestors
Methods
def execute(self, e: Executor) ‑> None
-
Expand source code Browse git
def execute(self, e: Executor) -> None: for check in self.checks: print(f"Running initialize() from {check}") check.start_initialize(e, self)
def join(self, e: Executor) ‑> None
-
Expand source code Browse git
def join(self, e: Executor) -> None: for check in self.checks: check.join_initialize(e)
class Manipulate (scenario: Scenario, phase: int | None = None, mz_service: str | None = None)
-
Expand source code Browse git
class Manipulate(Action): def __init__( self, scenario: "Scenario", phase: int | None = None, mz_service: str | None = None, ) -> None: assert phase is not None self.phase = phase - 1 self.mz_service = mz_service self.checks = scenario.check_objects def execute(self, e: Executor) -> None: assert self.phase is not None for check in self.checks: print(f"Running manipulate() from {check}") check.start_manipulate(e, self) def join(self, e: Executor) -> None: assert self.phase is not None for check in self.checks: check.join_manipulate(e, self)
Ancestors
Methods
def execute(self, e: Executor) ‑> None
-
Expand source code Browse git
def execute(self, e: Executor) -> None: assert self.phase is not None for check in self.checks: print(f"Running manipulate() from {check}") check.start_manipulate(e, self)
def join(self, e: Executor) ‑> None
-
Expand source code Browse git
def join(self, e: Executor) -> None: assert self.phase is not None for check in self.checks: check.join_manipulate(e, self)
class Sleep (interval: float)
-
Expand source code Browse git
class Sleep(Action): def __init__(self, interval: float) -> None: self.interval = interval def execute(self, e: Executor) -> None: print(f"Sleeping for {self.interval} seconds") time.sleep(self.interval) def join(self, e: Executor) -> None: pass
Ancestors
Methods
def execute(self, e: Executor) ‑> None
-
Expand source code Browse git
def execute(self, e: Executor) -> None: print(f"Sleeping for {self.interval} seconds") time.sleep(self.interval)
def join(self, e: Executor) ‑> None
-
Expand source code Browse git
def join(self, e: Executor) -> None: pass
class Testdrive (input: str, dedent: bool = True)
-
Expand source code Browse git
class Testdrive(Action): # Instruct pytest this class does not contain actual tests __test__ = False def __init__(self, input: str, dedent: bool = True) -> None: self.input = textwrap.dedent(input) if dedent else input self.handle: Any | None = None self.caller = getframeinfo(stack()[1][0]) def execute(self, e: Executor, mz_service: str | None = None) -> None: """Pass testdrive actions to be run by an Executor-specific implementation.""" self.handle = e.testdrive(self.input, self.caller, mz_service) def join(self, e: Executor) -> None: e.join(self.handle)
Ancestors
Methods
def execute(self, e: Executor, mz_service: str | None = None) ‑> None
-
Pass testdrive actions to be run by an Executor-specific implementation.
Expand source code Browse git
def execute(self, e: Executor, mz_service: str | None = None) -> None: """Pass testdrive actions to be run by an Executor-specific implementation.""" self.handle = e.testdrive(self.input, self.caller, mz_service)
def join(self, e: Executor) ‑> None
-
Expand source code Browse git
def join(self, e: Executor) -> None: e.join(self.handle)
class Validate (scenario: Scenario, mz_service: str | None = None)
-
Expand source code Browse git
class Validate(Action): def __init__(self, scenario: "Scenario", mz_service: str | None = None) -> None: self.checks = scenario.check_objects self.mz_service = mz_service def execute(self, e: Executor) -> None: for check in self.checks: print(f"Running validate() from {check}") check.start_validate(e, self) def join(self, e: Executor) -> None: for check in self.checks: check.join_validate(e)
Ancestors
Methods
def execute(self, e: Executor) ‑> None
-
Expand source code Browse git
def execute(self, e: Executor) -> None: for check in self.checks: print(f"Running validate() from {check}") check.start_validate(e, self)
def join(self, e: Executor) ‑> None
-
Expand source code Browse git
def join(self, e: Executor) -> None: for check in self.checks: check.join_validate(e)