Skip to content

Commit

Permalink
add v2 tests (#398)
Browse files Browse the repository at this point in the history 
<!--- Provide a general summary of your changes in the title above -->
<!--- Link the corresponding issues after you created the pull request
-->

## Types of changes
<!--- What types of changes does your code introduce? Put an `x` in all
the boxes that apply: -->
- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)

## Checklist:
<!--- Go over all the following points, and put an `x` in all the boxes
that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're
here to help! -->
- [ ] I have updated the [changelog](../CHANGELOG.md) accordingly.
- [ ] I have added tests to cover my changes.

---------

Co-authored-by: Deichmann <[email protected]>
Co-authored-by: Bernd Deichmann <[email protected]>
Co-authored-by: a-kleinf <[email protected]>
  • Loading branch information
@leon1995 @deichmab-draeger @a-kleinf
4 people authored Nov 18, 2024
1 parent a58d5f8 commit 463564c
Show file tree
Hide file tree
Showing 9 changed files with 2,216 additions and 45 deletions.
8 changes: 5 additions & 3 deletions .github/workflows/pat_integration.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,9 @@ on:
- master
- v*
jobs:
sdc11073_provider_v1:
sdc11073_provider_v2:
env:
EXPECTED_WAVEFORM_SAMPLES_4F: 100 # set to a low value as we cannot control GitHub ci network latency
strategy:
matrix:
os: [ ubuntu-latest, macos-latest, windows-latest ]
Expand All @@ -28,10 +30,10 @@ jobs:

- name: Run tests with tls enabled
if: ${{ matrix.tls_enable }}
run: python -m examples.ReferenceTest.run --tls
run: python -m examples.ReferenceTestV2.run --tls
timeout-minutes: 2

- name: Run tests with tls disabled
if: ${{ !matrix.tls_enable }}
run: python -m examples.ReferenceTest.run
run: python -m examples.ReferenceTestV2.run
timeout-minutes: 2
87 changes: 45 additions & 42 deletions examples/ReferenceTest/reference_consumer.py
View file
Original file line number Diff line number Diff line change
@@ -1,18 +1,19 @@
"""Reference test v1."""

import dataclasses
import enum
import os
import pathlib
import sys
import time
import traceback
import typing
import uuid
from collections import defaultdict
from concurrent import futures
from decimal import Decimal

import sdc11073.certloader
from sdc11073 import commlog, network
from sdc11073 import observableproperties
from sdc11073 import commlog, network, observableproperties
from sdc11073.certloader import mk_ssl_contexts_from_folder
from sdc11073.consumer import SdcConsumer
from sdc11073.definitions_sdc import SdcV1Definitions
Expand All @@ -24,7 +25,7 @@
ConsumerMdibMethods.DETERMINATIONTIME_WARN_LIMIT = 2.0

# ref_discovery_runs indicates the maximum executions of wsdiscovery search services, "0" -> run until service is found
discovery_runs = int(os.getenv('ref_discovery_runs', 0)) # noqa: SIM112
discovery_runs = int(os.getenv('ref_discovery_runs', "0")) # noqa: SIM112

ENABLE_COMMLOG = True

Expand Down Expand Up @@ -54,32 +55,36 @@ def get_epr() -> uuid.UUID:
return uuid.UUID(epr)
return uuid.UUID('12345678-6f55-11ea-9697-123456789abc')


class TestResult(enum.Enum):
"""
Represents the overall test result.
"""
"""Represents the overall test result."""

PASSED = 'PASSED'
FAILED = 'FAILED'

@dataclasses.dataclass
class TestCollector:
"""Test collector."""

overall_test_result: TestResult = TestResult.PASSED
test_messages: typing.List = dataclasses.field(default_factory=list)
test_messages: list = dataclasses.field(default_factory=list)

def add_result(self, test_step_message: str, test_step_result: TestResult):
"""Add result to result list."""
if not isinstance(test_step_result, TestResult):
raise ValueError("Unexpected parameter")
if self.overall_test_result is not TestResult.FAILED:
self.overall_test_result = test_step_result
self.test_messages.append(test_step_message)


def run_ref_test() -> TestCollector:
def run_ref_test() -> TestCollector: # noqa: PLR0915,PLR0912,C901
"""Run reference tests."""
test_collector = TestCollector()
adapter_ip = get_network_adapter().ip
print(f'using adapter address {adapter_ip}')
search_epr = get_epr()
print('Test step 1: discover device which endpoint ends with "{}"'.format(search_epr))
print(f'Test step 1: discover device which endpoint ends with "{search_epr}"')
wsd = WSDiscovery(str(adapter_ip))
wsd.start()
my_service = None
Expand All @@ -90,7 +95,7 @@ def run_ref_test() -> TestCollector:
for s in services:
if s.epr.endswith(str(search_epr)):
my_service = s
print('found service {}'.format(s.epr))
print(f'found service {s.epr}')
break
discovery_counter += 1
if discovery_runs and discovery_counter >= discovery_runs:
Expand All @@ -108,7 +113,7 @@ def run_ref_test() -> TestCollector:
client.start_all()
print('Test step 2 passed: connected to device')
test_collector.add_result('### Test 2 ### passed', TestResult.PASSED)
except:
except Exception: # noqa: BLE001
print(traceback.format_exc())
test_collector.add_result('### Test 2 ### failed', TestResult.FAILED)
return test_collector
Expand All @@ -120,7 +125,7 @@ def run_ref_test() -> TestCollector:
print('Test step 3&4 passed')
test_collector.add_result('### Test 3 ### passed', TestResult.PASSED)
test_collector.add_result('### Test 4 ### passed', TestResult.PASSED)
except:
except Exception: # noqa: BLE001
print(traceback.format_exc())
test_collector.add_result('### Test 3 ### failed', TestResult.FAILED)
test_collector.add_result('### Test 4 ### failed', TestResult.FAILED)
Expand All @@ -131,7 +136,7 @@ def run_ref_test() -> TestCollector:
print('Test step 5: check that at least one patient context exists')
patients = mdib.context_states.NODETYPE.get(pm.PatientContextState, [])
if len(patients) > 0:
print('found {} patients, Test step 5 passed'.format(len(patients)))
print(f'found {len(patients)} patients, Test step 5 passed')
test_collector.add_result('### Test 5 ### passed', TestResult.PASSED)
else:
print('found no patients, Test step 5 failed')
Expand All @@ -141,7 +146,7 @@ def run_ref_test() -> TestCollector:
print('Test step 6: check that at least one location context exists')
locations = mdib.context_states.NODETYPE.get(pm.LocationContextState, [])
if len(locations) > 0:
print('found {} locations, Test step 6 passed'.format(len(locations)))
print(f'found {len(locations)} locations, Test step 6 passed')
test_collector.add_result('### Test 6 ### passed', TestResult.PASSED)
else:
print('found no locations, Test step 6 failed')
Expand All @@ -151,22 +156,22 @@ def run_ref_test() -> TestCollector:
metric_updates = defaultdict(list)
alert_updates = defaultdict(list)

def onMetricUpdates(metricsbyhandle):
def _on_metric_updates(metricsbyhandle: dict):
print('onMetricUpdates', metricsbyhandle)
for k, v in metricsbyhandle.items():
metric_updates[k].append(v)

def onAlertUpdates(alertsbyhandle):
def _on_alert_updates(alertsbyhandle: dict):
print('onAlertUpdates', alertsbyhandle)
for k, v in alertsbyhandle.items():
alert_updates[k].append(v)

observableproperties.bind(mdib, metrics_by_handle=onMetricUpdates)
observableproperties.bind(mdib, alert_by_handle=onAlertUpdates)
observableproperties.bind(mdib, metrics_by_handle=_on_metric_updates)
observableproperties.bind(mdib, alert_by_handle=_on_alert_updates)

sleep_timer = 20
min_updates = sleep_timer // 5 - 1
print('will wait for {} seconds now, expecting at least {} updates per Handle'.format(sleep_timer, min_updates))
print(f'will wait for {sleep_timer} seconds now, expecting at least {min_updates} updates per Handle')
time.sleep(sleep_timer)
print(metric_updates)
print(alert_updates)
Expand All @@ -175,20 +180,20 @@ def onAlertUpdates(alertsbyhandle):
else:
for k, v in metric_updates.items():
if len(v) < min_updates:
print('found only {} updates for {}, test step 7 failed'.format(len(v), k))
print(f'found only {len(v)} updates for {k}, test step 7 failed')
test_collector.add_result(f'### Test 7 Handle {k} ### failed', TestResult.FAILED)
else:
print('found {} updates for {}, test step 7 ok'.format(len(v), k))
print(f'found {len(v)} updates for {k}, test step 7 ok')
test_collector.add_result(f'### Test 7 Handle {k} ### passed', TestResult.PASSED)
if len(alert_updates) == 0:
test_collector.add_result('### Test 8 ### failed', TestResult.FAILED)
else:
for k, v in alert_updates.items():
if len(v) < min_updates:
print('found only {} updates for {}, test step 8 failed'.format(len(v), k))
print(f'found only {len(v)} updates for {k}, test step 8 failed')
test_collector.add_result(f'### Test 8 Handle {k} ### failed', TestResult.FAILED)
else:
print('found {} updates for {}, test step 8 ok'.format(len(v), k))
print(f'found {len(v)} updates for {k}, test step 8 ok')
test_collector.add_result(f'### Test 8 Handle {k} ### passed', TestResult.PASSED)

print('Test step 9: call SetString operation')
Expand All @@ -201,28 +206,27 @@ def onAlertUpdates(alertsbyhandle):
for s in setstring_operations:
if s.Handle != setst_handle:
continue
print('setString Op ={}'.format(s))
print(f'setString Op ={s}')
try:
fut = client.set_service_client.set_string(s.Handle, 'hoppeldipop')
try:
res = fut.result(timeout=10)
print(res)
if res.InvocationInfo.InvocationState != InvocationState.FINISHED:
print('set string operation {} did not finish with "Fin":{}'.format(s.Handle, res))
print(f'set string operation {s.Handle} did not finish with "Fin":{res}')
test_collector.add_result('### Test 9(SetString) ### failed', TestResult.FAILED)
else:
print('set string operation {} ok:{}'.format(s.Handle, res))
print(f'set string operation {s.Handle} ok:{res}')
test_collector.add_result('### Test 9(SetString) ### passed', TestResult.PASSED)
except futures.TimeoutError:
print('timeout error')
test_collector.add_result('### Test 9(SetString) ### failed', TestResult.FAILED)
except Exception as ex:
except Exception as ex:# noqa: BLE001
print(f'Test 9(SetString): {ex}')
test_collector.add_result('### Test 9(SetString) ### failed', TestResult.FAILED)

print('Test step 9: call SetValue operation')
setvalue_operations = mdib.descriptions.NODETYPE.get(pm.SetValueOperationDescriptor, [])
# print('setvalue_operations', setvalue_operations)
setval_handle = 'numeric.ch0.vmd1_sco_0'
if len(setvalue_operations) == 0:
print('Test step 9 failed, no SetValue operation found')
Expand All @@ -231,22 +235,22 @@ def onAlertUpdates(alertsbyhandle):
for s in setvalue_operations:
if s.Handle != setval_handle:
continue
print('setNumericValue Op ={}'.format(s))
print(f'setNumericValue Op ={s}')
try:
fut = client.set_service_client.set_numeric_value(s.Handle, Decimal('42'))
try:
res = fut.result(timeout=10)
print(res)
if res.InvocationInfo.InvocationState != InvocationState.FINISHED:
print('set value operation {} did not finish with "Fin":{}'.format(s.Handle, res))
print(f'set value operation {s.Handle} did not finish with "Fin":{res}')
test_collector.add_result('### Test 9(SetValue) ### failed', TestResult.FAILED)
else:
print('set value operation {} ok:{}'.format(s.Handle, res))
print(f'set value operation {s.Handle} ok:{res}')
test_collector.add_result('### Test 9(SetValue) ### passed', TestResult.PASSED)
except futures.TimeoutError:
print('timeout error')
test_collector.add_result('### Test 9(SetValue) ### failed', TestResult.FAILED)
except Exception as ex:
except Exception as ex:# noqa: BLE001
print(f'Test 9(SetValue): {ex}')
test_collector.add_result('### Test 9(SetValue) ### failed', TestResult.FAILED)

Expand All @@ -260,27 +264,27 @@ def onAlertUpdates(alertsbyhandle):
for s in activate_operations:
if s.Handle != activate_handle:
continue
print('activate Op ={}'.format(s))
print(f'activate Op ={s}')
try:
fut = client.set_service_client.activate(s.Handle, 'hoppeldipop')
try:
res = fut.result(timeout=10)
print(res)
if res.InvocationInfo.InvocationState != InvocationState.FINISHED:
print('activate operation {} did not finish with "Fin":{}'.format(s.Handle, res))
print(f'activate operation {s.Handle} did not finish with "Fin":{res}')
test_collector.add_result('### Test 9(Activate) ### failed', TestResult.FAILED)
else:
print('activate operation {} ok:{}'.format(s.Handle, res))
print(f'activate operation {s.Handle} ok:{res}')
test_collector.add_result('### Test 9(Activate) ### passed', TestResult.PASSED)
except futures.TimeoutError:
print('timeout error')
test_collector.add_result('### Test 9(Activate) ### failed', TestResult.FAILED)
except Exception as ex:
except Exception as ex: # noqa: BLE001
print(f'Test 9(Activate): {ex}')
test_collector.add_result('### Test 9(Activate) ### failed', TestResult.FAILED)

print('Test step 10: cancel all subscriptions')
success = client._subscription_mgr.unsubscribe_all()
success = client._subscription_mgr.unsubscribe_all() # noqa: SLF001
if success:
test_collector.add_result('### Test 10(unsubscribe) ### passed', TestResult.PASSED)
else:
Expand All @@ -289,18 +293,17 @@ def onAlertUpdates(alertsbyhandle):
return test_collector

def main() -> TestCollector:
"""Execute reference tests."""
xtra_log_config = os.getenv('ref_xtra_log_cnf') # noqa: SIM112

import json
import logging.config

here = os.path.dirname(__file__)

with open(os.path.join(here, 'logging_default.json')) as f:
with pathlib.Path(__file__).parent.joinpath("logging_default.json").open() as f:
logging_setup = json.load(f)
logging.config.dictConfig(logging_setup)
if xtra_log_config is not None:
with open(xtra_log_config) as f:
with pathlib.Path(xtra_log_config).open() as f:
logging_setup2 = json.load(f)
logging.config.dictConfig(logging_setup2)
comm_logger = commlog.DirectoryLogger(log_folder=r'c:\temp\sdc_refclient_commlog',
Expand Down
Empty file added examples/ReferenceTestV2/__init__.py
View file
Empty file.
Loading

0 comments on commit 463564c

Please sign in to comment.