feat: testbed - Special Characteristics validation #48
Conversation
…t in the industry test cases
…alidator endpoint
mrpeterbasista
requested review from
dddenes and
matbmoser
and removed request for
matbmoser
ds-asmierzchalski
left a comment
ds-asmierzchalski
left a comment
There was a problem hiding this comment.
Hi @mrpeterbasista thanks for the contribution! I left some comments and would kindly ask you to have a look at them.
Furthermore I'm a little bit struggling with the workflow of this usecase. Is it really special characteristics specific or can it be used for other models as well?
You are getting information from an "events" field in the content. But the traceability open api specification doesn't specify this field.
If you have time we could also organize a small session to discuss those topics.
Kind regards
Adam
There was a problem hiding this comment.
Did you update every file which was importing the utils.py before to use base_utils.py now? It seems that at least traceability_test.py is missing.
There was a problem hiding this comment.
Hi Adam, not yet, I will merge main into our branch and need to check as the traceability_test.py was not part of the code while we developed SC.
| def validate_notification_payload(payload: Dict): | ||
| """ | ||
| Validate the structure, required fields, and formatting of a Catena-X | ||
| notification payload. | ||
|
|
||
| Steps performed: | ||
| 1. Ensure the presence of `header` and `content` sections. | ||
| 2. Validate mandatory header fields such as UUIDs, timestamps, and BPNs. | ||
| 3. Validate `listOfEvents` structure and required event fields. | ||
| 4. Accumulate and report all validation errors. | ||
|
|
||
| - :payload (Dict): Raw notification payload provided by the calling service. | ||
|
|
||
| return: a dict containing `{ "status": "ok" }` if validation succeeds, | ||
| otherwise an HTTPError is raised. | ||
| """ | ||
|
|
||
| errors: List[str] = [] | ||
|
|
||
| if 'header' not in payload or 'content' not in payload: | ||
| errors.append("Missing required sections: 'header' and/or 'content'.") | ||
| raise HTTPError( | ||
| Error.MISSING_REQUIRED_FIELD, | ||
| message='Required fields are missing in the notification', | ||
| details=errors) | ||
|
|
||
| header = payload.get('header', {}) | ||
| content = payload.get('content', {}) | ||
|
|
||
| required_header_fields = [ | ||
| 'messageId', 'context', 'sentDateTime', 'senderBpn', | ||
| 'receiverBpn', 'expectedResponseBy', 'relatedMessageId', 'version' | ||
| ] | ||
|
|
||
| for field in required_header_fields: | ||
| if field not in header: | ||
| errors.append(f'Missing header field: {field}') | ||
|
|
||
| uuid_pattern = re.compile(r'^(urn:uuid:)?[0-9a-fA-F-]{36}$') | ||
| bpn_pattern = re.compile(r'^BPN[LSA][A-Z0-9]{10}[A-Z0-9]{2}$') | ||
|
|
||
| for key in ['messageId', 'relatedMessageId']: | ||
| value = header.get(key) | ||
|
|
||
| if value and not uuid_pattern.match(value): | ||
| errors.append(f'Invalid UUID format in header.{key}: {value}') | ||
|
|
||
| for key in ['sentDateTime', 'expectedResponseBy']: | ||
| value = header.get(key) | ||
|
|
||
| try: | ||
| if value: | ||
| datetime.fromisoformat(value.replace('Z', '+00:00')) | ||
| except Exception: | ||
| errors.append(f'Invalid datetime format in header.{key}: {value}') | ||
|
|
||
| for key in ['senderBpn', 'receiverBpn']: | ||
| value = header.get(key) | ||
|
|
||
| if value and not bpn_pattern.match(value): | ||
| errors.append(f'Invalid BPN format in header.{key}: {value} (expected e.g. BPNL000000000000)') | ||
|
|
||
| if 'information' not in content or 'listOfEvents' not in content: | ||
| errors.append("Missing required content fields: 'information' and/or 'listOfEvents'.") | ||
| else: | ||
| list_of_events = content.get('listOfEvents', []) | ||
|
|
||
| if not isinstance(list_of_events, list) or not list_of_events: | ||
| errors.append('listOfEvents must be a non-empty array.') | ||
| else: | ||
| for i, event in enumerate(list_of_events): | ||
| for key in ['eventType', 'catenaXId', 'submodelSemanticId']: | ||
| if key not in event: | ||
| errors.append(f"Missing field '{key}' in listOfEvents[{i}]") | ||
|
|
||
| catena_id = event.get('catenaXId') | ||
|
|
||
| if catena_id and not uuid_pattern.match(catena_id): | ||
| errors.append(f'Invalid UUID format in listOfEvents[{i}].catenaXId: {catena_id}') | ||
|
|
||
| if errors: | ||
| logger.error(f'Notification validation failed: {errors}') | ||
| raise HTTPError( | ||
| Error.NOTIFICATION_VALIDATION_FAILED, | ||
| message='Notification validation failed', | ||
| details=errors) | ||
|
|
||
| return {'status': 'ok', | ||
| 'message': 'No errors found during validating the input json.'} |
There was a problem hiding this comment.
Is there a reason to validate the payload content manually instead of using a schema validator?
There was a problem hiding this comment.
Is this really just special characteristics specific? It looks like this could be used for another model, like the traction battery code, as well since the information is pulled from the payload itself. Please correct me if I'm wrong with that assumption.
3 participants
Description
The feature introduces Special Characteristics (SC) validation support in the Testbed. It enables receiving SC notifications from onboarding companies, fetching the corresponding Digital Twin from the DTR, and validating that SC aspects are correctly linked to the referenced Catena-X ID. The feature ensures that SC data structures, measurement types, and asset linkages comply with the semantic model.
Issues Linked
eclipse-tractusx/sig-release#1558
Pre-review checks
Please ensure to do as many of the following checks as possible, before asking for committer review: