pyopenadr/test/test_objects.py

from openleadr import objects, enums
from datetime import datetime, timedelta
from openleadr.utils import ensure_bytes
from openleadr.messaging import create_message, parse_message, validate_xml_schema
from pprint import pprint
import pytest


def test_oadr_event():
    event = objects.Event(
        event_descriptor=objects.EventDescriptor(
            event_id=1,
            modification_number=0,
            market_context='MarketContext1',
            event_status=enums.EVENT_STATUS.NEAR),
        active_period=objects.ActivePeriod(
            dtstart=datetime.now(),
            duration=timedelta(minutes=10)),
        event_signals=[objects.EventSignal(
            intervals=[
                objects.Interval(
                    dtstart=datetime.now(),
                    duration=timedelta(minutes=5),
                    uid=0,
                    signal_payload=1),
                objects.Interval(
                    dtstart=datetime.now(),
                    duration=timedelta(minutes=5),
                    uid=1,
                    signal_payload=2)],
            targets=[objects.Target(
                ven_id='1234'
            )],
            signal_name=enums.SIGNAL_NAME.LOAD_CONTROL,
            signal_type=enums.SIGNAL_TYPE.LEVEL,
            signal_id=1,
            current_value=0
        )],
        targets=[objects.Target(
            ven_id='1234'
        )]
    )

    response = objects.Response(response_code=200,
                                response_description='OK',
                                request_id='1234')
    msg = create_message('oadrDistributeEvent', response=response, events=[event])
    validate_xml_schema(ensure_bytes(msg))
    message_type, message_payload = parse_message(msg)


def test_oadr_event_targets_by_type():
    event = objects.Event(
        event_descriptor=objects.EventDescriptor(
            event_id=1,
            modification_number=0,
            market_context='MarketContext1',
            event_status=enums.EVENT_STATUS.NEAR),
        active_period=objects.ActivePeriod(
            dtstart=datetime.now(),
            duration=timedelta(minutes=10)),
        event_signals=[objects.EventSignal(
            intervals=[
                objects.Interval(
                    dtstart=datetime.now(),
                    duration=timedelta(minutes=5),
                    uid=0,
                    signal_payload=1),
                objects.Interval(
                    dtstart=datetime.now(),
                    duration=timedelta(minutes=5),
                    uid=1,
                    signal_payload=2)],
            targets=[objects.Target(
                ven_id='1234'
            )],
            signal_name=enums.SIGNAL_NAME.LOAD_CONTROL,
            signal_type=enums.SIGNAL_TYPE.LEVEL,
            signal_id=1,
            current_value=0
        )],
        targets_by_type={'ven_id': ['ven123']}
    )

    msg = create_message('oadrDistributeEvent', events=[event])
    validate_xml_schema(ensure_bytes(msg))
    message_type, message_payload = parse_message(msg)


def test_oadr_event_targets_and_targets_by_type():
    event = objects.Event(
        event_descriptor=objects.EventDescriptor(
            event_id=1,
            modification_number=0,
            market_context='MarketContext1',
            event_status=enums.EVENT_STATUS.NEAR),
        active_period=objects.ActivePeriod(
            dtstart=datetime.now(),
            duration=timedelta(minutes=10)),
        event_signals=[objects.EventSignal(
            intervals=[
                objects.Interval(
                    dtstart=datetime.now(),
                    duration=timedelta(minutes=5),
                    uid=0,
                    signal_payload=1),
                objects.Interval(
                    dtstart=datetime.now(),
                    duration=timedelta(minutes=5),
                    uid=1,
                    signal_payload=2)],
            targets=[objects.Target(
                ven_id='1234'
            )],
            signal_name=enums.SIGNAL_NAME.LOAD_CONTROL,
            signal_type=enums.SIGNAL_TYPE.LEVEL,
            signal_id=1,
            current_value=0
        )],
        targets=[{'ven_id': 'ven123'}],
        targets_by_type={'ven_id': ['ven123']}
    )

    msg = create_message('oadrDistributeEvent', events=[event])
    validate_xml_schema(ensure_bytes(msg))
    message_type, message_payload = parse_message(msg)


def test_oadr_event_targets_and_targets_by_type_invalid():
    with pytest.raises(ValueError):
        event = objects.Event(
            event_descriptor=objects.EventDescriptor(
                event_id=1,
                modification_number=0,
                market_context='MarketContext1',
                event_status=enums.EVENT_STATUS.NEAR),
            active_period=objects.ActivePeriod(
                dtstart=datetime.now(),
                duration=timedelta(minutes=10)),
            event_signals=[objects.EventSignal(
                intervals=[
                    objects.Interval(
                        dtstart=datetime.now(),
                        duration=timedelta(minutes=5),
                        uid=0,
                        signal_payload=1),
                    objects.Interval(
                        dtstart=datetime.now(),
                        duration=timedelta(minutes=5),
                        uid=1,
                        signal_payload=2)],
                targets=[objects.Target(
                    ven_id='1234'
                )],
                signal_name=enums.SIGNAL_NAME.LOAD_CONTROL,
                signal_type=enums.SIGNAL_TYPE.LEVEL,
                signal_id=1,
                current_value=0
            )],
            targets=[objects.Target(ven_id='ven456')],
            targets_by_type={'ven_id': ['ven123']}
        )

        msg = create_message('oadrDistributeEvent', events=[event])
        validate_xml_schema(ensure_bytes(msg))
        message_type, message_payload = parse_message(msg)


def test_oadr_event_no_targets():
    with pytest.raises(ValueError):
        event = objects.Event(
            event_descriptor=objects.EventDescriptor(
                event_id=1,
                modification_number=0,
                market_context='MarketContext1',
                event_status=enums.EVENT_STATUS.NEAR),
            active_period=objects.ActivePeriod(
                dtstart=datetime.now(),
                duration=timedelta(minutes=10)),
            event_signals=[objects.EventSignal(
                intervals=[
                    objects.Interval(
                        dtstart=datetime.now(),
                        duration=timedelta(minutes=5),
                        uid=0,
                        signal_payload=1),
                    objects.Interval(
                        dtstart=datetime.now(),
                        duration=timedelta(minutes=5),
                        uid=1,
                        signal_payload=2)],
                targets=[objects.Target(
                    ven_id='1234'
                )],
                signal_name=enums.SIGNAL_NAME.LOAD_CONTROL,
                signal_type=enums.SIGNAL_TYPE.LEVEL,
                signal_id=1,
                current_value=0
            )]
        )