Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def fake_filter(self, fake_dispatcher):
fake_filter = Mock()
fake_filter.filters = []
fake_filter.route = Mock(return_value=[fake_dispatcher])
fake_filter.get_type = Mock(return_value=Report)
return fake_filter
def actor(self, fake_db, buffer_size, delay):
report_model = Mock()
report_model.get_type = Mock(return_value=Report)
return PusherActor('pusher_test', report_model, fake_db, level_logger=logging.DEBUG, max_size=buffer_size,
delay=delay)
def setup(self):
self.push_socket = SafeContext.get_context().socket(zmq.PUSH)
self.push_socket.connect(self.addr)
self.add_handler(PoisonPillMessage, PoisonPillMessageHandler())
self.add_handler(StartMessage, StartHandler())
self.add_handler(Report, HWPCReportHandler(self.push_socket))
@pytest.fixture
def started_actor(self, init_actor, fake_db):
init_actor.send_control(StartMessage())
# remove OkMessage from control socket
_ = init_actor.receive_control(2000)
# remove 'connected' string from Queue
_ = fake_db.q.get(timeout=2)
return init_actor
def test_starting_actor_make_it_connect_to_database(self, init_actor, fake_db):
init_actor.send_control(StartMessage())
assert fake_db.q.get(timeout=2) == 'connected'
REPORT1 = Report(1, 2, 3)
REPORT2 = Report(3, 4, 5)
class FakeDB(BaseDB):
def __init__(self, content=[], *args, **kwargs):
self._content = content
self.q = Queue()
def connect(self):
self.q.put('connected', block=False)
def iter(self, report_model, stream_mode):
return self._content.__iter__()
def save(self, report, report_model):
self.q.put(report, block=False)
sub-report identifier is a tuple of two values, the first one is the *a*
value of the report, the second one is the *b* value or the *b2 value of the
report
"""
def __init__(self, primary=False):
DispatchRule.__init__(self, primary)
self.fields = ['A', 'B']
def get_formula_id(self, report):
b2_id = [] if report.b2 is None else [(report.a, report.b2)]
return [(report.a, report.b)] + b2_id
class Report2(Report):
""" Fake report that can contains two or three values : *a*, *c*, *c2* """
def __init__(self, a, c, c2=None):
self.a = a
self.c = c
self.c2 = c2
def __eq__(self, other):
if not isinstance(other, Report2):
return False
return other.a == self.a and other.c == self.c and other.c2 == self.c2
def __str__(self):
return '(' + str(self.a) + ',' + (str(self.c)
if self.c2 is None
else ('(' + self.c + ',' + self.c2 +
')')) + ')'
def setup(self):
self.add_handler(PoisonPillMessage, FakePoisonPillMessageHandler(self.state))
self.push_socket = SafeContext.get_context().socket(zmq.PUSH)
self.push_socket.connect(self.addr)
self.add_handler(Report, HWPCReportHandler(self.state, self.push_socket))
self.push_socket.send(pickle.dumps('created'))
def __init__(self, timestamp: datetime, sensor: str, target: str, power: float, metadata: Dict[str, Any]):
"""
Initialize a Power report using the given parameters.
:param datetime timestamp: Report timestamp
:param str sensor: Sensor name
:param str target: Target name
:param float power: Power value
:param dict metadata: Metadata values, can be anything that add useful information
"""
Report.__init__(self, timestamp, sensor, target)
self.metadata = metadata
self.power = power
def __init__(self, timestamp: datetime, sensor: str, target: str, metadata: Dict[str, Any]):
"""
Initialize a Power report using the given parameters.
:param timestamp: Report timestamp
:param sensor: Sensor name
:param target: Target name
:param metadata: Metadata values, can be anything that add useful information
"""
Report.__init__(self, timestamp, sensor, target)
self.metadata = metadata
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
from __future__ import annotations
from datetime import datetime
from typing import Dict, Any
from powerapi.report import Report
class FormulaReport(Report):
"""
FormulaReport stores information about a formula.
This is useful to gather information about a running formula in order to debug or compute statistics.
"""
def __init__(self, timestamp: datetime, sensor: str, target: str, metadata: Dict[str, Any]):
"""
Initialize a Power report using the given parameters.
:param timestamp: Report timestamp
:param sensor: Sensor name
:param target: Target name
:param metadata: Metadata values, can be anything that add useful information
"""
Report.__init__(self, timestamp, sensor, target)
self.metadata = metadata
def setup(self):
"""
Check if there is a primary group by rule. Set define
StartMessage, PoisonPillMessage and Report handlers
"""
Actor.setup(self)
if self.state.route_table.primary_dispatch_rule is None:
raise NoPrimaryDispatchRuleRuleException()
self.add_handler(Report, FormulaDispatcherReportHandler(self.state))
self.add_handler(PoisonPillMessage, DispatcherPoisonPillMessageHandler(self.state))
self.add_handler(StartMessage, StartHandler(self.state))