Clean up unused utils, added more tests

Signed-off-by: Stan Janssen <stan.janssen@elaad.nl>

openleadr/objects.py

@@ -92,7 +92,7 @@ class Target:
 
 @dataclass
 class EventDescriptor:
-    event_id: int
+    event_id: str
     modification_number: int
     market_context: str
     event_status: str

openleadr/utils.py

@@ -19,7 +19,6 @@ from dataclasses import is_dataclass, asdict
 from collections import OrderedDict
 from openleadr import enums, objects
 import asyncio
-import itertools
 import re
 import ssl
 import hashlib
@@ -39,24 +38,6 @@ def generate_id(*args, **kwargs):
     return str(uuid.uuid4())
 
 
-def indent_xml(message):
-    """
-    Indents the XML in a nice way.
-    """
-    INDENT_SIZE = 2
-    lines = [line.strip() for line in message.split("\n") if line.strip() != ""]
-    indent = 0
-    for i, line in enumerate(lines):
-        if i == 0:
-            continue
-        if re.search(r'^</[^>]+>$', line):
-            indent = indent - INDENT_SIZE
-        lines[i] = " " * indent + line
-        if not (re.search(r'</[^>]+>$', line) or line.endswith("/>")):
-            indent = indent + INDENT_SIZE
-    return "\n".join(lines)
-
-
 def flatten_xml(message):
     """
     Flatten the entire XML structure.
@@ -316,18 +297,6 @@ def parse_boolean(value):
         return False
 
 
-def peek(iterable):
-    """
-    Peek into an iterable.
-    """
-    try:
-        first = next(iterable)
-    except StopIteration:
-        return None
-    else:
-        return itertools.chain([first], iterable)
-
-
 def datetimeformat(value, format=DATETIME_FORMAT):
     """
     Format a given datetime as a UTC ISO3339 string.
@@ -507,13 +476,6 @@ def get_cert_fingerprint_from_request(request):
             return certificate_fingerprint_from_der(der_bytes)
 
 
-def get_certificate_common_name(request):
-    cert = request.transport.get_extra_info('peercert')
-    if cert:
-        subject = dict(x[0] for x in cert['subject'])
-        return subject.get('commonName')
-
-
 def group_targets_by_type(list_of_targets):
     targets_by_type = {}
     for target in list_of_targets:

test/test_objects.py

@@ -1,5 +1,5 @@
 from openleadr import objects, enums
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from openleadr.utils import ensure_bytes
 from openleadr.messaging import create_message, parse_message, validate_xml_schema
 from pprint import pprint
@@ -198,3 +198,32 @@ def test_oadr_event_no_targets():
                 current_value=0
             )]
         )
+
+def test_event_signal_with_grouped_targets():
+    event_signal = objects.EventSignal(intervals=[objects.Interval(dtstart=datetime.now(timezone.utc),
+                                                                  duration=timedelta(minutes=10),
+                                                                  signal_payload=1)],
+                                       signal_name='simple',
+                                       signal_type='level',
+                                       signal_id='signal123',
+                                       targets_by_type={'ven_id': ['ven123', 'ven456']})
+    assert event_signal.targets == [objects.Target(ven_id='ven123'), objects.Target(ven_id='ven456')]
+
+def test_event_signal_with_incongruent_targets():
+    with pytest.raises(ValueError):
+        event_signal = objects.EventSignal(intervals=[objects.Interval(dtstart=datetime.now(timezone.utc),
+                                                                      duration=timedelta(minutes=10),
+                                                                      signal_payload=1)],
+                                           signal_name='simple',
+                                           signal_type='level',
+                                           signal_id='signal123',
+                                           targets=[objects.Target(ven_id='ven123')],
+                                           targets_by_type={'ven_id': ['ven123', 'ven456']})
+
+
+def test_event_descriptor_modification_number():
+    event_descriptor = objects.EventDescriptor(event_id='event123',
+                                               modification_number=None,
+                                               market_context='http://marketcontext01',
+                                               event_status='near')
+    assert event_descriptor.modification_number == 0

test/test_schema.py

@@ -14,7 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from openleadr.utils import generate_id, indent_xml, ensure_bytes
+from openleadr.utils import generate_id, ensure_bytes
 from openleadr.messaging import create_message, validate_xml_schema
 from openleadr import enums
 from lxml import etree

test/test_utils.py

@@ -153,5 +153,6 @@ def test_get_event_from_deque():
     assert d.popleft() is msg_one
     assert d.popleft() is msg_two
     assert d.popleft() is msg_three
+    assert utils.get_next_event_from_deque(d) is None