How to use powerapi - 10 common examples

def test_init_fields_name_test():
    """
    test if the field's names of the dispatch_rule identifier tuple are
    correctly initialized
    """
    assert HWPCDispatchRule(HWPCDepthLevel.TARGET).fields == ['target']
    assert HWPCDispatchRule(HWPCDepthLevel.ROOT).fields == ['sensor']
    assert HWPCDispatchRule(HWPCDepthLevel.SOCKET).fields == ['sensor',
                                                              'socket']
    assert HWPCDispatchRule(HWPCDepthLevel.CORE).fields == ['sensor', 'socket',
                                                            'core']

def test_init_fields_name_test():
    """
    test if the field's names of the dispatch_rule identifier tuple are
    correctly initialized
    """
    assert HWPCDispatchRule(HWPCDepthLevel.TARGET).fields == ['target']
    assert HWPCDispatchRule(HWPCDepthLevel.ROOT).fields == ['sensor']
    assert HWPCDispatchRule(HWPCDepthLevel.SOCKET).fields == ['sensor',
                                                              'socket']
    assert HWPCDispatchRule(HWPCDepthLevel.CORE).fields == ['sensor', 'socket',
                                                            'core']

def test_init_fields_name_test():
    """
    test if the field's names of the dispatch_rule identifier tuple are
    correctly initialized
    """
    assert HWPCDispatchRule(HWPCDepthLevel.TARGET).fields == ['target']
    assert HWPCDispatchRule(HWPCDepthLevel.ROOT).fields == ['sensor']
    assert HWPCDispatchRule(HWPCDepthLevel.SOCKET).fields == ['sensor',
                                                              'socket']
    assert HWPCDispatchRule(HWPCDepthLevel.CORE).fields == ['sensor', 'socket',
                                                            'core']

def test_cant_convert_to_type():
    """
    add an argument that must catch an int value, Parse a string that
    contains only this argument with a value that is not an int test if an
    """
    parser = MainParser(help_arg=False)
    parser.add_argument('a', type=int)

    with pytest.raises(BadTypeException):
        parser.parse('-a a'.split())

def test_default_type():
    """
    add an argument without specifing the type it must catch. Parse a string
    that contains only this argument and test if the value contained in the
    result is a string

    """
    parser = MainParser(help_arg=False)
    parser.add_argument('a')
    result = parser.parse('-a 1'.split())
    assert len(result) == 1
    assert 'a' in result
    assert isinstance(result['a'], str)

# Setup signal handler
        def term_handler(_, __):
            puller.hard_kill()
            dispatcher.hard_kill()
            pusher.hard_kill()
            exit(0)

        signal.signal(signal.SIGTERM, term_handler)
        signal.signal(signal.SIGINT, term_handler)

        stream_mode = True
        supervisor = BackendSupervisor(stream_mode)

        # Pusher
        output_mongodb = MongoDB(DB_URI, 'MongoDB1', 'test_result')
        pusher = PusherActor("pusher_mongodb", PowerModel(), output_mongodb, level_logger=LOG_LEVEL)

        # Formula
        formula_factory = (lambda name, verbose:
                           DummyFormulaActor(name, {'id_pusher': pusher}, level_logger=verbose))

        # Dispatcher
        route_table = RouteTable()
        route_table.dispatch_rule(HWPCReport, HWPCDispatchRule(getattr(HWPCDepthLevel, 'ROOT'), primary=True))

        dispatcher = DispatcherActor('dispatcher', formula_factory, route_table,
                                     level_logger=LOG_LEVEL)

        # Puller
        input_mongodb = MongoDB(DB_URI, 'MongoDB1', 'test_hwrep')
        report_filter = Filter()
        report_filter.filter(lambda msg: True, dispatcher)

def test_csvdb_second_rapl_missing(self, csvdb):
        """
        Create two reports, one is full, second without rapl, then return None
        """
        csvdb.add_files(SECOND_RAPL_MISSING)
        csvdb.connect()
        group_name = [path.split('/')[-1][:-4] for path in SECOND_RAPL_MISSING]

        csvdb_iter = csvdb.iter(HWPCModel(), False)
        for i in range(2):
            report = next(csvdb_iter)
            for group in group_name:
                if i == 1 and "rapl" in group:
                    assert group not in report.groups
                else:
                    assert group in report.groups
        with pytest.raises(StopIteration) as pytest_wrapped:
            next(csvdb_iter)
        assert pytest_wrapped.type == StopIteration

def test_csvdb_second_primary_missing(self, csvdb):
        """
        Create one full HWPCReport (the first), then return None
        """
        csvdb.add_files(SECOND_PRIMARY_MISSING)
        csvdb.connect()
        group_name = [path.split('/')[-1][:-4] for path in SECOND_PRIMARY_MISSING]

        csvdb_iter = csvdb.iter(HWPCModel(), False)
        report = next(csvdb_iter)
        for group in group_name:
            assert group in report.groups
        assert report.timestamp == timestamp_to_datetime(1539260664189)
        with pytest.raises(StopIteration) as pytest_wrapped:
            next(csvdb_iter)
        assert pytest_wrapped.type == StopIteration

def test_csvdb_first_primary_missing(self, csvdb):
        """
        Create one full HWPCReport (the second), then return None
        """
        csvdb.add_files(FIRST_PRIMARY_MISSING)
        csvdb.connect()
        group_name = [path.split('/')[-1][:-4] for path in FIRST_PRIMARY_MISSING]

        csvdb_iter = csvdb.iter(HWPCModel(), False)
        report = next(csvdb_iter)
        for group in group_name:
            assert group in report.groups
        assert report.timestamp == timestamp_to_datetime(1539260665189)
        with pytest.raises(StopIteration) as pytest_wrapped:
            next(csvdb_iter)
        assert pytest_wrapped.type == StopIteration

def test_csvdb_first_rapl_missing(self, csvdb):
        """
        Create two reports, one without rapl, second is full, then return None
        """
        csvdb.add_files(FIRST_RAPL_MISSING)
        csvdb.connect()
        group_name = [path.split('/')[-1][:-4] for path in FIRST_RAPL_MISSING]

        csvdb_iter = csvdb.iter(HWPCModel(), False)
        for i in range(2):
            report = next(csvdb_iter)
            for group in group_name:
                if i == 0 and "rapl" in group:
                    assert group not in report.groups
                else:
                    assert group in report.groups
        with pytest.raises(StopIteration) as pytest_wrapped:
            next(csvdb_iter)
        assert pytest_wrapped.type == StopIteration