AAS HTTP Client Documentation
Loading...
Searching...
No Matches
test_wrapper.py
Go to the documentation of this file.
1from http import client
2import pytest
3from pathlib import Path
4from aas_http_client.classes.wrapper.sdk_wrapper import IdEncoding, Level, create_by_config, SdkWrapper, create_by_dict, create_by_url
5from basyx.aas import model
6import aas_http_client.utilities.model_builder as model_builder
8from urllib.parse import urlparse
9import json
10from aas_http_client.utilities import encoder
11import random
12
13JAVA_SERVER_PORTS = [8075]
14PYTHON_SERVER_PORTS = [5080, 80]
15DOTNET_SERVER_PORTS = [5043]
16
17AIMC_SM_ID = "https://fluid40.de/ids/sm/7644_4034_2556_2369"
18SM_ID = "fluid40/sm_http_client_unit_tests"
19SHELL_ID = "fluid40/aas_http_client_unit_tests"
20
21CONFIG_FILES = [
22 "./tests/server_configs/test_java_server_config.yml",
23 "./tests/server_configs/test_dotnet_server_config.yml",
24 "./tests/server_configs/test_python_server_config.yml"
25]
26
27# CONFIG_FILES = [
28# "./tests/server_configs/test_dotnet_server_config.yml",
29# ]
30
31@pytest.fixture(params=CONFIG_FILES, scope="module")
32def wrapper(request) -> SdkWrapper:
33 try:
34 file = Path(request.param).resolve()
35
36 if not file.exists():
37 raise FileNotFoundError(f"Configuration file {file} does not exist.")
38
39 wrapper = create_by_config(file, basic_auth_password="")
40
41 rand = random.randint(0, 10)
42 if (rand % 2) == 0:
43 wrapper.set_encoded_ids(IdEncoding.encoded)
44
45 wrapper.set_encoded_ids(IdEncoding.encoded)
46
47 except Exception as e:
48 raise RuntimeError("Unable to connect to server.")
49
50 shells = wrapper.get_all_asset_administration_shells()
51 if shells is None:
52 raise RuntimeError("No shells found on server. Please check the server configuration.")
53
54 return wrapper
55
56@pytest.fixture(scope="module")
57def shared_sme_string() -> model.Property:
58 # create a Submodel
59 return model_builder.create_base_submodel_element_property("sme_property_string", model.datatypes.String, "Sample String Value")
60
61@pytest.fixture(scope="module")
62def shared_sme_bool() -> model.Property:
63 # create a Submodel
64 return model_builder.create_base_submodel_element_property("sme_property_bool", model.datatypes.Boolean, True)
65
66@pytest.fixture(scope="module")
67def shared_sme_int() -> model.Property:
68 # create a Submodel
69 return model_builder.create_base_submodel_element_property("sme_property_int", model.datatypes.Integer, 262)
70
71@pytest.fixture(scope="module")
72def shared_sme_float() -> model.Property:
73 # create a Submodel
74 return model_builder.create_base_submodel_element_property("sme_property_float", model.datatypes.Float, 262.3)
75
76@pytest.fixture(scope="module")
77def shared_sm() -> model.Submodel:
78 # create a Submodel
79 submodel = model_builder.create_base_submodel(identifier=SM_ID, id_short="sm_http_client_unit_tests")
80 submodel.category = "Unit Test"
81 return submodel
82
83@pytest.fixture(scope="module")
84def shared_aas(shared_sm: model.Submodel) -> model.AssetAdministrationShell:
85 # create an AAS
86 aas = model_builder.create_base_aas(identifier=SHELL_ID, id_short="aas_http_client_unit_tests")
87
88 # add Submodel to AAS
89 sdk_tools.add_submodel_to_aas(aas, shared_sm)
90
91 return aas
92
93def test_000a_create_wrapper_by_url(wrapper: SdkWrapper):
94 base_url: str = wrapper.base_url
95 new_client: SdkWrapper = create_by_url(base_url=base_url)
96 assert new_client is not None
97
98def test_000b_create_wrapper_by_dict(wrapper: SdkWrapper):
99 base_url: str = wrapper.base_url
100
101 config_dict: dict = {
102 "BaseUrl": base_url
103 }
104
105 new_client: SdkWrapper = create_by_dict(configuration=config_dict)
106 assert new_client is not None
107
108def test_000c_get_client(wrapper: SdkWrapper):
109 client = wrapper.get_client()
110 assert client is not None
111 root = client.get_root()
112 assert root is not None
113
114def test_001a_connect(wrapper: SdkWrapper):
115 assert wrapper is not None
116
118 result = wrapper.get_all_asset_administration_shells()
119 assert result is not None
120
121 for shell in result.results:
122 shell_id = shell.id
123
124 if wrapper.get_encoded_ids() == IdEncoding.encoded:
125 shell_id = encoder.encode_base_64(shell_id)
126
127 if shell_id:
128 delete_result = wrapper.delete_asset_administration_shell_by_id(shell_id)
129 assert delete_result
130
131 shells_result = wrapper.get_all_asset_administration_shells()
132 assert len(shells_result.results) == 0
133
134def test_001c_delete_all_submodels(wrapper: SdkWrapper):
135 result = wrapper.get_all_submodels()
136 assert result is not None
137
138 for submodel in result.results:
139 submodel_id = submodel.id
140
141 if wrapper.get_encoded_ids() == IdEncoding.encoded:
142 submodel_id = encoder.encode_base_64(submodel_id)
143
144 if submodel_id:
145 delete_result = wrapper.delete_submodel_by_id(submodel_id)
146 assert delete_result
147
148 submodels_result = wrapper.get_all_submodels()
149 assert len(submodels_result.results) == 0
150
152 shells = wrapper.get_all_asset_administration_shells()
153 assert shells is not None
154 assert len(shells.results) == 0
155
156def test_003_post_asset_administration_shell(wrapper: SdkWrapper, shared_aas: model.AssetAdministrationShell):
157 shell = wrapper.post_asset_administration_shell(shared_aas)
158
159 assert shell is not None
160 assert shell.id == SHELL_ID
161 assert shell.id_short == shared_aas.id_short
162
163 shells = wrapper.get_all_asset_administration_shells()
164 assert shells is not None
165 assert len(shells.results) == 1
166 assert shells.results[0].id_short == shared_aas.id_short
167 assert shells.results[0].id == SHELL_ID
168
169def test_004a_get_asset_administration_shell_by_id(wrapper: SdkWrapper, shared_aas: model.AssetAdministrationShell):
170 shell_id = SHELL_ID
171
172 if wrapper.get_encoded_ids() == IdEncoding.encoded:
173 shell_id = encoder.encode_base_64(shell_id)
174
175 shell = wrapper.get_asset_administration_shell_by_id(shell_id)
176
177 assert shell is not None
178 assert shell.id_short == shared_aas.id_short
179 assert shell.id == SHELL_ID
180
182 shell = wrapper.get_asset_administration_shell_by_id("non_existent_id")
183
184 assert shell is None
185
186def test_005a_put_asset_administration_shell_by_id(wrapper: SdkWrapper, shared_aas: model.AssetAdministrationShell):
187 aas = model.AssetAdministrationShell(id_=shared_aas.asset_information.global_asset_id, asset_information=shared_aas.asset_information)
188 aas.id_short = shared_aas.id_short
189
190 description_text = "Put description for unit tests"
191 aas.description = model.MultiLanguageTextType({"en": description_text})
192 aas.submodel = shared_aas.submodel # Keep existing submodels
193
194 shell_id = SHELL_ID
195
196 if wrapper.get_encoded_ids() == IdEncoding.encoded:
197 shell_id = encoder.encode_base_64(shell_id)
198
199 result = wrapper.put_asset_administration_shell_by_id(shell_id, aas)
200
201 assert result
202
203 shell = wrapper.get_asset_administration_shell_by_id(shell_id)
204
205 assert shell is not None
206 assert shell.id_short == shared_aas.id_short
207 assert shell.id == SHELL_ID
208 # description must have changed
209 assert shell.description.get("en", "") == description_text
210 assert shell.description.get("en", "") != shared_aas.description.get("en", "")
211 # submodels must be retained
212 assert len(shell.submodel) == len(shared_aas.submodel)
213
214 # The display name must be empty
215 # currently not working in dotnet
216 # assert len(get_result.get("displayName", {})) == 0
217
218 # # restore to its original state
219 wrapper.put_asset_administration_shell_by_id(shell_id, shared_aas) # Restore original submodel
220
221def test_005b_put_asset_administration_shell_by_id(wrapper: SdkWrapper, shared_aas: model.AssetAdministrationShell):
222 # put with other ID
223 id_short = "put_short_id"
224 identifier = f"fluid40/{id_short}"
225 asset_info = model_builder.create_base_asset_information(identifier)
226 aas = model.AssetAdministrationShell(id_=asset_info.global_asset_id, asset_information=asset_info)
227 aas.id_short = id_short
228
229 description_text = {"en": "Updated description for unit tests"}
230 aas.description = model.MultiLanguageTextType(description_text)
231
232 shell_id = SHELL_ID
233
234 if wrapper.get_encoded_ids() == IdEncoding.encoded:
235 shell_id = encoder.encode_base_64(SHELL_ID)
236
237 parsed = urlparse(wrapper.base_url)
238 if int(parsed.port) in PYTHON_SERVER_PORTS:
239 # NOTE: Python server crashes by this test
240 result = False
241 else:
242 result = wrapper.put_asset_administration_shell_by_id(shell_id, aas)
243
244 assert not result
245
246 shell = wrapper.get_asset_administration_shell_by_id(shell_id)
247 assert shell.description.get("en", "") != description_text
248 assert shell.description.get("en", "") == shared_aas.description.get("en", "")
249
251 shell_id = SHELL_ID
252
253 if wrapper.get_encoded_ids() == IdEncoding.encoded:
254 shell_id = encoder.encode_base_64(SHELL_ID)
255
256 reference = wrapper.get_asset_administration_shell_by_id_reference_aas_repository(shell_id)
257
258 assert reference is not None
259 assert len(reference.key) == 1
260 assert reference.key[0].value == SHELL_ID
261
262def test_007_get_submodel_by_id_aas_repository(wrapper: SdkWrapper, shared_sm: model.Submodel):
263 shell_id = SHELL_ID
264
265 if wrapper.get_encoded_ids() == IdEncoding.encoded:
266 shell_id = encoder.encode_base_64(SHELL_ID)
267
268 submodel = wrapper.get_submodel_by_id_aas_repository(shell_id, shared_sm.id)
269
270 assert submodel is None
271
272def test_008_get_all_submodels(wrapper: SdkWrapper):
273 submodels = wrapper.get_all_submodels()
274 assert submodels is not None
275 assert len(submodels.results) == 0
276
277def test_009a_post_submodel(wrapper: SdkWrapper, shared_sm: model.Submodel):
278 submodel = wrapper.post_submodel(shared_sm)
279
280 assert submodel is not None
281 assert submodel.id == SM_ID
282 assert submodel.id_short == shared_sm.id_short
283
284 submodels = wrapper.get_all_submodels()
285 assert submodels is not None
286 assert len(submodels.results) == 1
287 assert submodels.results[0].id_short == shared_sm.id_short
288 assert submodels.results[0].id == SM_ID
289
290def test_009b_post_submodel(wrapper: SdkWrapper):
291 sm_template_file = Path(f"./tests/test_data/aimc.json").resolve()
292
293 with Path.open(sm_template_file, "r", encoding="utf-8") as f:
294 sm_data = json.load(f)
295
296 submodel = sdk_tools.convert_to_object(sm_data)
297
298 result = wrapper.post_submodel(submodel)
299
300 assert result is not None
301 assert result.id == AIMC_SM_ID
302
303 get_result = wrapper.get_all_submodels()
304 assert get_result is not None
305 submodels = get_result.results
306 assert len(submodels) == 2
307
308def test_010_get_submodel_by_id_aas_repository(wrapper: SdkWrapper, shared_sm: model.Submodel):
309 shell_id = SHELL_ID
310 sm_id = SM_ID
311
312 if wrapper.get_encoded_ids() == IdEncoding.encoded:
313 shell_id = encoder.encode_base_64(SHELL_ID)
314 sm_id = encoder.encode_base_64(SM_ID)
315
316 submodel = wrapper.get_submodel_by_id_aas_repository(shell_id, sm_id)
317
318 parsed = urlparse(wrapper.base_url)
319 if int(parsed.port) in JAVA_SERVER_PORTS:
320 # Basyx java server do not provide this endpoint
321 assert submodel is None
322 else:
323 assert submodel is not None
324 assert submodel.id_short == shared_sm.id_short
325 assert submodel.id == SM_ID
326
327def test_011a_get_submodel_by_id(wrapper: SdkWrapper, shared_sm: model.Submodel):
328 sm_id = SM_ID
329
330 if wrapper.get_encoded_ids() == IdEncoding.encoded:
331 sm_id = encoder.encode_base_64(SM_ID)
332
333 submodel = wrapper.get_submodel_by_id(sm_id)
334
335 assert submodel is not None
336 assert submodel.id_short == shared_sm.id_short
337 assert submodel.id == SM_ID
338
339def test_011b_get_submodel_by_id(wrapper: SdkWrapper):
340 result = wrapper.get_submodel_by_id("non_existent_id")
341
342 assert result is None
343
344def test_011c_get_submodel_by_id(wrapper: SdkWrapper):
345 sm_id = AIMC_SM_ID
346
347 if wrapper.get_encoded_ids() == IdEncoding.encoded:
348 sm_id = encoder.encode_base_64(AIMC_SM_ID)
349
350 result = wrapper.get_submodel_by_id(sm_id)
351
352 assert result is not None
353 assert result.id == AIMC_SM_ID
354
355def test_011d_get_submodel_by_id(wrapper: SdkWrapper):
356 sm_id = AIMC_SM_ID
357
358 if wrapper.get_encoded_ids() == IdEncoding.encoded:
359 sm_id = encoder.encode_base_64(AIMC_SM_ID)
360
361 result = wrapper.get_submodel_by_id(sm_id, level=Level.core)
362
363 assert result is not None
364 assert result.id == AIMC_SM_ID
365 # assert "submodelElements" not in result
366
367def test_012_patch_submodel_by_id(wrapper: SdkWrapper, shared_sm: model.Submodel):
368 sm = model.Submodel(shared_sm.id_short)
369 sm.id_short = shared_sm.id_short
370
371 description_text = "Patched description for unit tests"
372 sm.description = model.MultiLanguageTextType({"en": description_text})
373
374 sm_id = SM_ID
375
376 if wrapper.get_encoded_ids() == IdEncoding.encoded:
377 sm_id = encoder.encode_base_64(SM_ID)
378
379 result = wrapper.patch_submodel_by_id(sm_id, sm)
380
381 parsed = urlparse(wrapper.base_url)
382 if int(parsed.port) in JAVA_SERVER_PORTS or int(parsed.port) in PYTHON_SERVER_PORTS:
383 # Basyx java server do not provide this endpoint
384 assert not result
385 else:
386 assert result
387
388 submodel = wrapper.get_submodel_by_id(sm_id)
389 assert submodel is not None
390 assert submodel.id_short == shared_sm.id_short
391 assert submodel.id == SM_ID
392 # Only the description may change in patch.
393 assert submodel.description.get("en", "") == description_text
394 assert submodel.description.get("en", "") != shared_sm.description.get("en", "")
395 # The display name must remain the same.
396 assert submodel.display_name == shared_sm.display_name
397 assert len(submodel.submodel_element) == len(shared_sm.submodel_element)
398
399def test_013_put_submodel_by_id_aas_repository(wrapper: SdkWrapper, shared_sm: model.Submodel):
400 sm = model.Submodel(SM_ID)
401 sm.id_short = shared_sm.id_short
402
403 description_text = "Put via shell description for unit tests"
404 sm.description = model.MultiLanguageTextType({"en": description_text})
405 sm.display_name = shared_sm.display_name # Keep existing display name because of problems with empty lists
406
407 shell_id = SHELL_ID
408 sm_id = SM_ID
409
410 if wrapper.get_encoded_ids() == IdEncoding.encoded:
411 shell_id = encoder.encode_base_64(SHELL_ID)
412 sm_id = encoder.encode_base_64(SM_ID)
413
414 result = wrapper.put_submodel_by_id_aas_repository(shell_id, sm_id, sm)
415
416 parsed = urlparse(wrapper.base_url)
417 if int(parsed.port) in JAVA_SERVER_PORTS:
418 # Basyx java server do not provide this endpoint
419 assert not result
420 else:
421 assert result
422
423 submodel = wrapper.get_submodel_by_id_aas_repository(shell_id, sm_id)
424 assert submodel is not None
425 assert submodel.id_short == shared_sm.id_short
426 assert submodel.id == SM_ID
427 # description must have changed
428 assert submodel.description.get("en", "") == description_text
429 assert submodel.description.get("en", "") != shared_sm.description.get("en", "")
430 # display name stays
431 assert submodel.display_name == shared_sm.display_name
432 # category was not set an must be empty
433 assert submodel.category is None
434 assert len(submodel.submodel_element) == 0
435
436 # restore to its original state
437 wrapper.put_submodel_by_id_aas_repository(shell_id, sm_id, shared_sm) # Restore original submodel
438
439def test_014_put_submodels_by_id(wrapper: SdkWrapper, shared_sm: model.Submodel):
440 sm = model.Submodel(shared_sm.id)
441 sm.id_short = shared_sm.id_short
442
443 description_text = "Put description for unit tests"
444 sm.description = model.MultiLanguageTextType({"en": description_text})
445 sm.display_name = shared_sm.display_name # Keep existing display name because of problems with empty lists
446
447 sm_id = SM_ID
448
449 if wrapper.get_encoded_ids() == IdEncoding.encoded:
450 sm_id = encoder.encode_base_64(SM_ID)
451
452 result = wrapper.put_submodels_by_id(sm_id, sm)
453
454 assert result
455
456 submodel = wrapper.get_submodel_by_id(sm_id)
457 assert submodel is not None
458 assert submodel.id_short == shared_sm.id_short
459 assert submodel.id == SM_ID
460 # description must have changed
461 assert submodel.description.get("en", "") == description_text
462 assert submodel.description.get("en", "") != shared_sm.description.get("en", "")
463 # display name stays
464 # assert submodel.display_name == shared_sm.display_name
465 # category was not set an must be empty
466 assert submodel.category is None
467 assert len(submodel.submodel_element) == 0
468
469 # restore to its original state
470 wrapper.put_submodels_by_id(sm_id, shared_sm) # Restore original submodel
471
473 sm_id = SM_ID
474
475 if wrapper.get_encoded_ids() == IdEncoding.encoded:
476 sm_id = encoder.encode_base_64(SM_ID)
477
478 submodel_elements = wrapper.get_all_submodel_elements_submodel_repository(sm_id)
479
480 assert submodel_elements is not None
481 assert len(submodel_elements.results) == 0
482
483def test_016a_post_submodel_element_submodel_repo(wrapper: SdkWrapper, shared_sme_string: model.Property):
484 sm_id = SM_ID
485
486 if wrapper.get_encoded_ids() == IdEncoding.encoded:
487 sm_id = encoder.encode_base_64(SM_ID)
488
489 submodel_element = wrapper.post_submodel_element_submodel_repo(sm_id, shared_sme_string)
490
491 assert submodel_element is not None
492
493 assert isinstance(submodel_element, model.Property)
494 property: model.Property = submodel_element
495 assert property.value == shared_sme_string.value
496
497 assert submodel_element.id_short == shared_sme_string.id_short
498 assert submodel_element.description.get("en", "") == shared_sme_string.description.get("en", "")
499 assert submodel_element.display_name.get("en", "") == shared_sme_string.display_name.get("en", "")
500 assert submodel_element.value == shared_sme_string.value
501
502 submodel_elements = wrapper.get_all_submodel_elements_submodel_repository(sm_id)
503
504 assert submodel_elements is not None
505 assert len(submodel_elements.results) == 1
506
507def test_016b_post_submodel_element_submodel_repo(wrapper: SdkWrapper, shared_sme_bool: model.Property):
508 sm_id = SM_ID
509
510 if wrapper.get_encoded_ids() == IdEncoding.encoded:
511 sm_id = encoder.encode_base_64(SM_ID)
512
513 submodel_element = wrapper.post_submodel_element_submodel_repo(sm_id, shared_sme_bool)
514
515 assert submodel_element is not None
516
517 assert isinstance(submodel_element, model.Property)
518 property: model.Property = submodel_element
519 assert property.value == shared_sme_bool.value
520
521 assert submodel_element.id_short == shared_sme_bool.id_short
522 assert submodel_element.description.get("en", "") == shared_sme_bool.description.get("en", "")
523 assert submodel_element.display_name.get("en", "") == shared_sme_bool.display_name.get("en", "")
524 assert submodel_element.value == shared_sme_bool.value
525
526 submodel_elements = wrapper.get_all_submodel_elements_submodel_repository(sm_id)
527
528 assert submodel_elements is not None
529 assert len(submodel_elements.results) == 2
530
531def test_016c_post_submodel_element_submodel_repo(wrapper: SdkWrapper, shared_sme_int: model.Property):
532 sm_id = SM_ID
533
534 if wrapper.get_encoded_ids() == IdEncoding.encoded:
535 sm_id = encoder.encode_base_64(SM_ID)
536
537 submodel_element = wrapper.post_submodel_element_submodel_repo(sm_id, shared_sme_int)
538
539 assert submodel_element is not None
540
541 assert isinstance(submodel_element, model.Property)
542 property: model.Property = submodel_element
543 assert property.value == shared_sme_int.value
544
545 assert submodel_element.id_short == shared_sme_int.id_short
546 assert submodel_element.description.get("en", "") == shared_sme_int.description.get("en", "")
547 assert submodel_element.display_name.get("en", "") == shared_sme_int.display_name.get("en", "")
548 assert submodel_element.value == shared_sme_int.value
549
550 submodel_elements = wrapper.get_all_submodel_elements_submodel_repository(sm_id)
551
552 assert submodel_elements is not None
553 assert len(submodel_elements.results) == 3
554
555def test_016d_post_submodel_element_submodel_repo(wrapper: SdkWrapper, shared_sme_float: model.Property):
556 sm_id = SM_ID
557
558 if wrapper.get_encoded_ids() == IdEncoding.encoded:
559 sm_id = encoder.encode_base_64(SM_ID)
560
561 submodel_element = wrapper.post_submodel_element_submodel_repo(sm_id, shared_sme_float)
562
563 assert submodel_element is not None
564
565 assert isinstance(submodel_element, model.Property)
566 property: model.Property = submodel_element
567 assert property.value == shared_sme_float.value
568
569 assert submodel_element.id_short == shared_sme_float.id_short
570 assert submodel_element.description.get("en", "") == shared_sme_float.description.get("en", "")
571 assert submodel_element.display_name.get("en", "") == shared_sme_float.display_name.get("en", "")
572 assert submodel_element.value == shared_sme_float.value
573
574 submodel_elements = wrapper.get_all_submodel_elements_submodel_repository(sm_id)
575
576 assert submodel_elements is not None
577 assert len(submodel_elements.results) == 4
578
579def test_017a_get_submodel_element_by_path_submodel_repo(wrapper: SdkWrapper, shared_sme_string: model.Property):
580 sm_id = SM_ID
581
582 if wrapper.get_encoded_ids() == IdEncoding.encoded:
583 sm_id = encoder.encode_base_64(SM_ID)
584
585 submodel_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_string.id_short)
586
587 assert submodel_element is not None
588
589 assert isinstance(submodel_element, model.Property)
590
591 assert submodel_element.id_short == shared_sme_string.id_short
592 assert submodel_element.description.get("en", "") == shared_sme_string.description.get("en", "")
593 assert submodel_element.display_name.get("en", "") == shared_sme_string.display_name.get("en", "")
594 assert submodel_element.value == shared_sme_string.value
595
596def test_017b_get_submodel_element_by_path_submodel_repo(wrapper: SdkWrapper, shared_sme_bool: model.Property):
597 sm_id = SM_ID
598
599 if wrapper.get_encoded_ids() == IdEncoding.encoded:
600 sm_id = encoder.encode_base_64(SM_ID)
601
602 submodel_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_bool.id_short)
603
604 assert submodel_element is not None
605
606 assert isinstance(submodel_element, model.Property)
607
608 assert submodel_element.id_short == shared_sme_bool.id_short
609 assert submodel_element.description.get("en", "") == shared_sme_bool.description.get("en", "")
610 assert submodel_element.display_name.get("en", "") == shared_sme_bool.display_name.get("en", "")
611 assert submodel_element.value == shared_sme_bool.value
612
613def test_017c_get_submodel_element_by_path_submodel_repo(wrapper: SdkWrapper, shared_sme_int: model.Property):
614 sm_id = SM_ID
615
616 if wrapper.get_encoded_ids() == IdEncoding.encoded:
617 sm_id = encoder.encode_base_64(SM_ID)
618
619 submodel_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_int.id_short)
620
621 assert submodel_element is not None
622
623 assert isinstance(submodel_element, model.Property)
624
625 assert submodel_element.id_short == shared_sme_int.id_short
626 assert submodel_element.description.get("en", "") == shared_sme_int.description.get("en", "")
627 assert submodel_element.display_name.get("en", "") == shared_sme_int.display_name.get("en", "")
628 assert submodel_element.value == shared_sme_int.value
629
630def test_017d_get_submodel_element_by_path_submodel_repo(wrapper: SdkWrapper, shared_sme_float: model.Property):
631 sm_id = SM_ID
632
633 if wrapper.get_encoded_ids() == IdEncoding.encoded:
634 sm_id = encoder.encode_base_64(SM_ID)
635
636 submodel_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_float.id_short)
637
638 assert submodel_element is not None
639
640 assert isinstance(submodel_element, model.Property)
641
642 assert submodel_element.id_short == shared_sme_float.id_short
643 assert submodel_element.description.get("en", "") == shared_sme_float.description.get("en", "")
644 assert submodel_element.display_name.get("en", "") == shared_sme_float.display_name.get("en", "")
645 assert submodel_element.value == shared_sme_float.value
646
647def test_018a_patch_submodel_element_by_path_value_only_submodel_repo(wrapper: SdkWrapper, shared_sme_string: model.Property):
648 new_value = "Patched String Value"
649
650 sm_id = SM_ID
651
652 if wrapper.get_encoded_ids() == IdEncoding.encoded:
653 sm_id = encoder.encode_base_64(SM_ID)
654
655 submodel_element: model.Property = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_string.id_short)
656 old_value = submodel_element.value
657
658 result = wrapper.patch_submodel_element_by_path_value_only_submodel_repo(sm_id, shared_sme_string.id_short, new_value)
659
660 parsed = urlparse(wrapper.base_url)
661 if int(parsed.port) in PYTHON_SERVER_PORTS:
662 # NOTE: python server do not provide this endpoint
663 assert result is False
664 else:
665 assert result is True
666
667 submodel_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_string.id_short)
668
669 assert submodel_element is not None
670 assert submodel_element.id_short == shared_sme_string.id_short
671 assert submodel_element.description.get("en", "") == shared_sme_string.description.get("en", "")
672 assert submodel_element.display_name.get("en", "") == shared_sme_string.display_name.get("en", "")
673
674 assert isinstance(submodel_element, model.Property)
675 property: model.Property = submodel_element
676 assert property.value == new_value
677 assert property.value != old_value
678
679def test_018b_patch_submodel_element_by_path_value_only_submodel_repo(wrapper: SdkWrapper, shared_sme_bool: model.Property):
680 new_value = "false"
681
682 sm_id = SM_ID
683
684 if wrapper.get_encoded_ids() == IdEncoding.encoded:
685 sm_id = encoder.encode_base_64(SM_ID)
686
687 submodel_element: model.Property = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_bool.id_short)
688 old_value = submodel_element.value
689
690 result = wrapper.patch_submodel_element_by_path_value_only_submodel_repo(sm_id, shared_sme_bool.id_short, new_value)
691
692 parsed = urlparse(wrapper.base_url)
693 if int(parsed.port) in PYTHON_SERVER_PORTS:
694 # NOTE: python server do not provide this endpoint
695 assert result is False
696 else:
697 assert result is True
698
699 submodel_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_bool.id_short)
700
701 assert submodel_element is not None
702 assert submodel_element.id_short == shared_sme_bool.id_short
703 assert submodel_element.description.get("en", "") == shared_sme_bool.description.get("en", "")
704 assert submodel_element.display_name.get("en", "") == shared_sme_bool.display_name.get("en", "")
705
706 assert isinstance(submodel_element, model.Property)
707 property: model.Property = submodel_element
708 assert property.value == json.loads(new_value)
709 assert property.value != old_value
710
711def test_018c_patch_submodel_element_by_path_value_only_submodel_repo(wrapper: SdkWrapper, shared_sme_int: model.Property):
712 new_value = "263"
713
714 sm_id = SM_ID
715
716 if wrapper.get_encoded_ids() == IdEncoding.encoded:
717 sm_id = encoder.encode_base_64(SM_ID)
718
719 submodel_element: model.Property = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_int.id_short)
720 old_value = submodel_element.value
721
722 result = wrapper.patch_submodel_element_by_path_value_only_submodel_repo(sm_id, shared_sme_int.id_short, new_value)
723
724 parsed = urlparse(wrapper.base_url)
725 if int(parsed.port) in PYTHON_SERVER_PORTS:
726 # NOTE: python server do not provide this endpoint
727 assert result is False
728 else:
729 assert result is True
730
731 submodel_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_int.id_short)
732
733 assert submodel_element is not None
734 assert submodel_element.id_short == shared_sme_int.id_short
735 assert submodel_element.description.get("en", "") == shared_sme_int.description.get("en", "")
736 assert submodel_element.display_name.get("en", "") == shared_sme_int.display_name.get("en", "")
737
738 assert isinstance(submodel_element, model.Property)
739 property: model.Property = submodel_element
740 assert property.value == int(new_value)
741 assert property.value != old_value
742
743def test_018d_patch_submodel_element_by_path_value_only_submodel_repo(wrapper: SdkWrapper, shared_sme_float: model.Property):
744 new_value = "262.1"
745
746 sm_id = SM_ID
747
748 if wrapper.get_encoded_ids() == IdEncoding.encoded:
749 sm_id = encoder.encode_base_64(SM_ID)
750
751 submodel_element: model.Property = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_float.id_short)
752 old_value = submodel_element.value
753
754 result = wrapper.patch_submodel_element_by_path_value_only_submodel_repo(sm_id, shared_sme_float.id_short, new_value)
755
756 parsed = urlparse(wrapper.base_url)
757 if int(parsed.port) in PYTHON_SERVER_PORTS:
758 # NOTE: python server do not provide this endpoint
759 assert result is False
760 else:
761 assert result is True
762
763 submodel_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_float.id_short)
764
765 assert submodel_element is not None
766 assert submodel_element.id_short == shared_sme_float.id_short
767 assert submodel_element.description.get("en", "") == shared_sme_float.description.get("en", "")
768 assert submodel_element.display_name.get("en", "") == shared_sme_float.display_name.get("en", "")
769
770 assert isinstance(submodel_element, model.Property)
771 property: model.Property = submodel_element
772 assert property.value == float(new_value)
773 assert property.value != old_value
774 assert property.value == 262.1
775
777 submodel_element_list = model.SubmodelElementList(id_short="sme_list_1", type_value_list_element=model.Property, value_type_list_element=model.datatypes.String)
778
779 sm_id = SM_ID
780
781 if wrapper.get_encoded_ids() == IdEncoding.encoded:
782 sm_id = encoder.encode_base_64(SM_ID)
783
784 post_list_element_result = wrapper.post_submodel_element_submodel_repo(sm_id, submodel_element_list)
785
786 assert post_list_element_result is not None
787
788 property = model_builder.create_base_submodel_element_property(None, model.datatypes.String, "Value in List")# idShort must be empty for list elements
789
790 result = wrapper.post_submodel_element_by_path_submodel_repo(sm_id, submodel_element_list.id_short, property)
791
792 assert result is not None
793 assert result.id_short == property.id_short
794
795 submodel = wrapper.get_submodel_by_id(sm_id)
796
797 assert submodel is not None
798 elements = submodel.submodel_element
799 assert len(elements) == 5 # 4 previous properties + 1 list
800 element = list(elements)[4]
801 assert element is not None
802 assert isinstance(element, model.SubmodelElementList)
803
804 assert element.id_short == submodel_element_list.id_short
805 list_elements = element.value
806 assert len(list_elements) == 1
807 list_element = list(list_elements)[0]
808 assert isinstance(list_element, model.Property)
809
810 assert "hack" in list_element.id_short
811 assert list_element.value == property.value
812
814 submodel_element_collection = model.SubmodelElementCollection(id_short="sme_collection_1")
815
816 sm_id = SM_ID
817
818 if wrapper.get_encoded_ids() == IdEncoding.encoded:
819 sm_id = encoder.encode_base_64(SM_ID)
820
821 first_result = wrapper.post_submodel_element_submodel_repo(sm_id, submodel_element_collection)
822
823 assert first_result is not None
824
825 property = model_builder.create_base_submodel_element_property("sme_property_in_collection", model.datatypes.String, "Value in List")
826 result = wrapper.post_submodel_element_by_path_submodel_repo(sm_id, submodel_element_collection.id_short, property)
827
828 assert result is not None
829 assert result.id_short== property.id_short
830
831 submodel = wrapper.get_submodel_by_id(sm_id)
832
833 assert submodel is not None
834 elements = submodel.submodel_element
835 assert len(elements) == 6
836 assert list(elements)[5].id_short == submodel_element_collection.id_short
837 list_elements = list(elements)[5].value
838 assert len(list_elements) == 1
839 assert list(list_elements)[0].id_short == property.id_short
840 assert list(list_elements)[0].value == property.value
841
842 base_url: str = wrapper.base_url
843 new_wrapper = create_by_url(base_url=base_url)
844 assert new_wrapper is not None
845
846 sm = new_wrapper.get_submodel_by_id(AIMC_SM_ID)
847 assert sm is None
848
849 decoded_id = encoder.encode_base_64(AIMC_SM_ID)
850 decoded_sm = new_wrapper.get_submodel_by_id(decoded_id)
851 assert decoded_sm is not None
852 assert decoded_sm.id == AIMC_SM_ID
853
854def test_020a_encoded_ids(wrapper: SdkWrapper):
855 base_url: str = wrapper.base_url
856 new_wrapper: SdkWrapper = create_by_url(base_url=base_url)
857 assert new_wrapper is not None
858
859 sm = new_wrapper.get_submodel_by_id(AIMC_SM_ID)
860 assert sm is None
861
862 encoded_id = encoder.encode_base_64(AIMC_SM_ID)
863 encoded_sm = new_wrapper.get_submodel_by_id(encoded_id)
864 assert encoded_sm is not None
865 assert encoded_sm.id == AIMC_SM_ID
866
867def test_020b_encoded_ids(wrapper: SdkWrapper):
868 base_url: str = wrapper.base_url
869 new_wrapper: SdkWrapper = create_by_url(base_url=base_url)
870 assert new_wrapper is not None
871
872 sm = new_wrapper.get_asset_administration_shell_by_id(SHELL_ID)
873 assert sm is None
874
875 encoded_id = encoder.encode_base_64(SHELL_ID)
876 encoded_sm = new_wrapper.get_asset_administration_shell_by_id(encoded_id)
877 assert encoded_sm is not None
878 assert encoded_sm.id == SHELL_ID
879
881 parsed = urlparse(wrapper.base_url)
882 if int(parsed.port) in JAVA_SERVER_PORTS or int(parsed.port) in PYTHON_SERVER_PORTS:
883 # NOTE: python server implementation differs
884 # NOTE: Basyx java server do not provide this endpoint
885 return
886
887 file_sme = model.File("file_sme", content_type="application/pdf")
888
889 sm_id = SM_ID
890
891 if wrapper.get_encoded_ids() == IdEncoding.encoded:
892 sm_id = encoder.encode_base_64(SM_ID)
893
894 file_post_result = wrapper.post_submodel_element_submodel_repo(sm_id, file_sme)
895 assert file_post_result is not None
896
897 filename = "https.pdf"
898 file = Path(f"./tests/test_data/{filename}").resolve()
899 result = wrapper.experimental_post_file_by_path_submodel_repo(sm_id, file_sme.id_short, file)
900 assert result is True
901
902 result_sme = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, file_sme.id_short)
903
904 assert result_sme is not None
905 assert result_sme.id_short == file_sme.id_short
906
907 assert result_sme.content_type == file_sme.content_type
908 assert result_sme.value == f"/{filename}"
909
911 parsed = urlparse(wrapper.base_url)
912 if int(parsed.port) in JAVA_SERVER_PORTS or int(parsed.port) in PYTHON_SERVER_PORTS:
913 # NOTE: python server implementation differs
914 # NOTE: Basyx java server do not provide this endpoint
915 return
916
917 sm_id = SM_ID
918
919 if wrapper.get_encoded_ids() == IdEncoding.encoded:
920 sm_id = encoder.encode_base_64(SM_ID)
921
922 attachment = wrapper.experimental_get_file_by_path_submodel_repo(sm_id, "file_sme")
923 assert attachment is not None
924 assert attachment.content_type == "application/pdf"
925 assert isinstance(attachment.content, bytes)
926 assert len(attachment.content) > 0
927 assert attachment.content.startswith(b"%PDF-1.7")
928 assert attachment.filename == "/https.pdf"
929
931 parsed = urlparse(wrapper.base_url)
932 if int(parsed.port) in JAVA_SERVER_PORTS or int(parsed.port) in PYTHON_SERVER_PORTS:
933 # NOTE: python server implementation differs
934 # NOTE: Basyx java server do not provide this endpoint
935 return
936
937 filename = "aimc.json"
938 file = Path(f"./tests/test_data/{filename}").resolve()
939
940 sm_id = SM_ID
941
942 if wrapper.get_encoded_ids() == IdEncoding.encoded:
943 sm_id = encoder.encode_base_64(SM_ID)
944
945 result = wrapper.experimental_put_file_by_path_submodel_repo(sm_id, "file_sme", file)
946 assert result is True
947
948 get_result = wrapper.experimental_get_file_by_path_submodel_repo(sm_id, "file_sme")
949 assert get_result is not None
950 assert len(get_result.content) > 0
951 assert get_result.content.startswith(b"{\n")
952 assert get_result.filename == f"/{filename}"
953 assert get_result.content_type == "application/json"
954
955 result_sme = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, "file_sme")
956 assert result_sme is not None
957 assert result_sme.value == f"/{filename}"
958
960 parsed = urlparse(wrapper.base_url)
961 if int(parsed.port) in JAVA_SERVER_PORTS or int(parsed.port) in PYTHON_SERVER_PORTS:
962 # NOTE: python server implementation differs
963 # NOTE: Basyx java server do not provide this endpoint
964 return
965
966 sm_id = SM_ID
967
968 if wrapper.get_encoded_ids() == IdEncoding.encoded:
969 sm_id = encoder.encode_base_64(SM_ID)
970
971 result = wrapper.experimental_delete_file_by_path_submodel_repo(sm_id, "file_sme")
972 assert result is True
973
974 get_result = wrapper.experimental_get_file_by_path_submodel_repo(sm_id, "file_sme")
975 assert get_result is None
976
977 result_sme = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, "file_sme")
978 assert result_sme is not None
979 assert result_sme.value == None
980
982 parsed = urlparse(wrapper.base_url)
983 if int(parsed.port) in PYTHON_SERVER_PORTS:
984 # NOTE: python server implementation differs
985 return
986
987 shell_id = SHELL_ID
988
989 if wrapper.get_encoded_ids() == IdEncoding.encoded:
990 shell_id = encoder.encode_base_64(SHELL_ID)
991
992 result = wrapper.get_thumbnail_aas_repository(shell_id)
993 assert result is None
994
996 parsed = urlparse(wrapper.base_url)
997 if int(parsed.port) in PYTHON_SERVER_PORTS:
998 # NOTE: python server implementation differs
999 return
1000
1001 shell_id = SHELL_ID
1002
1003 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1004 shell_id = encoder.encode_base_64(SHELL_ID)
1005
1006 filename = "Pen_Machine.png"
1007 file = Path(f"./tests/test_data/{filename}").resolve()
1008
1009 result = wrapper.put_thumbnail_aas_repository(shell_id, file.name, file)
1010 assert result is True
1011
1013 parsed = urlparse(wrapper.base_url)
1014 if int(parsed.port) in PYTHON_SERVER_PORTS:
1015 # NOTE: python server implementation differs
1016 return
1017
1018 shell_id = SHELL_ID
1019
1020 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1021 shell_id = encoder.encode_base_64(SHELL_ID)
1022
1023 result = wrapper.get_thumbnail_aas_repository(shell_id)
1024 assert result is not None
1025
1026 assert len(result.content) > 0
1027 assert result.content.startswith(b"\x89PNG\r\n\x1a\n")
1028 assert result.filename == "thumbnail"
1029 assert result.content_type == "image/png"
1030
1032 parsed = urlparse(wrapper.base_url)
1033 if int(parsed.port) in PYTHON_SERVER_PORTS:
1034 # NOTE: python server do not provide this endpoint
1035 return
1036
1037 shell_id = SHELL_ID
1038
1039 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1040 shell_id = encoder.encode_base_64(SHELL_ID)
1041
1042 result = wrapper.delete_thumbnail_aas_repository(shell_id)
1043 assert result is True
1044
1045 get_result = wrapper.get_thumbnail_aas_repository(shell_id)
1046 assert get_result is None
1047
1049 shell_id = SHELL_ID
1050
1051 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1052 shell_id = encoder.encode_base_64(SHELL_ID)
1053
1054 result = wrapper.get_all_submodel_references_aas_repository(shell_id)
1055 assert result is not None
1056 references = result.results
1057 assert len(references) == 1
1058
1060 shell_id = SHELL_ID
1061
1062 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1063 shell_id = encoder.encode_base_64(SHELL_ID)
1064
1065 id = "temp_sm_id"
1066 id_short = "TempSM"
1067 temp_sml_ref = model.ModelReference.from_referable(model_builder.create_base_submodel(identifier=id, id_short=id_short))
1068
1069 result = wrapper.post_submodel_reference_aas_repository(shell_id, temp_sml_ref)
1070
1071 assert result is not None
1072 assert len(result.get("keys", [])) > 0
1073 key: dict = result.get("keys", [])[0]
1074 assert key.get("value", "") == id
1075 assert key.get("type", "") == "Submodel"
1076
1077 check_result = wrapper.get_all_submodel_references_aas_repository(shell_id)
1078 assert check_result is not None
1079 assert len(check_result.results) == 2
1080
1082 shell_id = SHELL_ID
1083 sm_id = "temp_sm_id"
1084
1085 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1086 shell_id = encoder.encode_base_64(SHELL_ID)
1087 sm_id = encoder.encode_base_64(sm_id)
1088
1089 result = wrapper.delete_submodel_reference_by_id_aas_repository(shell_id, sm_id)
1090
1091 assert result is True
1092
1093 get_result = wrapper.get_all_submodel_references_aas_repository(shell_id)
1094 assert get_result is not None
1095 assert len(get_result.results) == 1
1096
1097def test_032_put_submodel_element_by_path_submodel_repo(wrapper: SdkWrapper, shared_sme_string: model.Property):
1098 sm_id = SM_ID
1099
1100 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1101 sm_id = encoder.encode_base_64(SM_ID)
1102
1103 sme: model.Property = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_string.id_short)
1104 old_value = sme.value
1105
1106 new_value = "New Value via PUT"
1107 shared_sme_string.value = new_value
1108
1109 result = wrapper.put_submodel_element_by_path_submodel_repo(sm_id, shared_sme_string.id_short, shared_sme_string)
1110
1111 assert result is True
1112
1113 sme = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_string.id_short)
1114
1115 assert sme is not None
1116 assert sme.id_short == shared_sme_string.id_short
1117 assert sme.value == new_value
1118 assert sme.value != old_value
1119 assert sme.description.get("en", "") == shared_sme_string.description.get("en", "")
1120 assert sme.display_name.get("en", "") == shared_sme_string.display_name.get("en", "")
1121
1122 # restore original value
1123 shared_sme_string.value = "Sample String Value"
1124 wrapper.put_submodel_element_by_path_submodel_repo(sm_id, shared_sme_string.id_short, shared_sme_string)
1125
1126def test_033a_get_submodel_element_by_path_value_only_submodel_repo(wrapper: SdkWrapper, shared_sme_string: model.Property):
1127 sm_id = SM_ID
1128
1129 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1130 sm_id = encoder.encode_base_64(SM_ID)
1131
1132 value = wrapper.get_submodel_element_by_path_value_only_submodel_repo(sm_id, shared_sme_string.id_short)
1133
1134 parsed = urlparse(wrapper.base_url)
1135 if parsed.port in PYTHON_SERVER_PORTS:
1136 # NOTE: python server do not provide this endpoint
1137 assert value is None
1138 return
1139
1140 assert value is not None
1141
1142 submodel_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_string.id_short)
1143 assert submodel_element is not None
1144 prop: model.Property = submodel_element
1145 assert value == prop.value
1146
1147def test_033b_get_submodel_element_by_path_value_only_submodel_repo(wrapper: SdkWrapper, shared_sme_int: model.Property):
1148 sm_id = SM_ID
1149
1150 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1151 sm_id = encoder.encode_base_64(SM_ID)
1152
1153 value = wrapper.get_submodel_element_by_path_value_only_submodel_repo(sm_id, shared_sme_int.id_short)
1154
1155 parsed = urlparse(wrapper.base_url)
1156 if parsed.port in PYTHON_SERVER_PORTS:
1157 # NOTE: python server do not provide this endpoint
1158 assert value is None
1159 return
1160
1161 assert value is not None
1162
1163 submodel_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_int.id_short)
1164 assert submodel_element is not None
1165 prop: model.Property = submodel_element
1166 assert int(value) == prop.value
1167
1168def test_033c_get_submodel_element_by_path_value_only_submodel_repo(wrapper: SdkWrapper, shared_sme_float: model.Property):
1169 sm_id = SM_ID
1170
1171 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1172 sm_id = encoder.encode_base_64(SM_ID)
1173
1174 value = wrapper.get_submodel_element_by_path_value_only_submodel_repo(sm_id, shared_sme_float.id_short)
1175
1176 parsed = urlparse(wrapper.base_url)
1177 if parsed.port in PYTHON_SERVER_PORTS:
1178 # NOTE: python server do not provide this endpoint
1179 assert value is None
1180 return
1181
1182 assert value is not None
1183
1184 submodel_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_float.id_short)
1185 assert submodel_element is not None
1186 prop: model.Property = submodel_element
1187 assert float(value) == prop.value
1188
1189def test_033d_get_submodel_element_by_path_value_only_submodel_repo(wrapper: SdkWrapper, shared_sme_bool: model.Property):
1190 sm_id = SM_ID
1191
1192 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1193 sm_id = encoder.encode_base_64(SM_ID)
1194
1195 value = wrapper.get_submodel_element_by_path_value_only_submodel_repo(sm_id, shared_sme_bool.id_short)
1196
1197 parsed = urlparse(wrapper.base_url)
1198 if parsed.port in PYTHON_SERVER_PORTS:
1199 # NOTE: python server do not provide this endpoint
1200 assert value is None
1201 return
1202
1203 assert value is not None
1204
1205 sm_data = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_bool.id_short)
1206 assert sm_data is not None
1207 #assert bool(value) == bool(sm_data.get("value", ""))
1208
1209def test_034_get_submodel_by_id_value_only(wrapper: SdkWrapper, shared_sm: model.Submodel):
1210 sm_id = SM_ID
1211
1212 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1213 sm_id = encoder.encode_base_64(SM_ID)
1214
1215 response = wrapper.get_submodel_by_id_value_only(sm_id)
1216
1217 parsed = urlparse(wrapper.base_url)
1218 if parsed.port in PYTHON_SERVER_PORTS:
1219 # NOTE: python server do not provide this endpoint
1220 assert response is None
1221 return
1222 elif parsed.port in DOTNET_SERVER_PORTS:
1223 assert response is not None
1224 value = response[shared_sm.id_short]
1225 else:
1226 assert response is not None
1227 value = response
1228
1229 assert value is not None
1230 assert len(value) > 3
1231 assert "sme_property_int" in value
1232 assert int(value["sme_property_int"]) == 263
1233 assert "sme_property_string" in value
1234 assert value["sme_property_string"] == "Sample String Value"
1235 assert "sme_property_float" in value
1236 assert float(value["sme_property_float"]) == 262.1
1237
1238def test_035_patch_submodel_by_id_value_only(wrapper: SdkWrapper, shared_sm: model.Submodel, shared_sme_string: model.Property, shared_sme_int: model.Property, shared_sme_float: model.Property):
1239 sm_id = SM_ID
1240
1241 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1242 sm_id = encoder.encode_base_64(SM_ID)
1243
1244 value_dict = {
1245 shared_sme_string.id_short: shared_sme_string.value,
1246 shared_sme_int.id_short: str(shared_sme_int.value),
1247 shared_sme_float.id_short: str(shared_sme_float.value)
1248 }
1249
1250 # patch_dict = {shared_sm.id: value_dict}
1251
1252 patch_dict = value_dict
1253
1254 parsed = urlparse(wrapper.base_url)
1255 if parsed.port in PYTHON_SERVER_PORTS:
1256 # NOTE: python server do not provide this endpoint
1257 return
1258
1259 if parsed.port in JAVA_SERVER_PORTS:
1260 # NOTE: java server endpoint seems to work not correctly
1261 return
1262
1263 elif parsed.port in DOTNET_SERVER_PORTS:
1264 patch_dict = value_dict
1265
1266 result = wrapper.patch_submodel_by_id_value_only(sm_id, patch_dict)
1267
1268 assert result is True
1269
1270 string_prop_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_string.id_short)
1271 int_prop_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_int.id_short)
1272 float_prop_element = wrapper.get_submodel_element_by_path_submodel_repo(sm_id, shared_sme_float.id_short)
1273
1274 assert string_prop_element is not None
1275 assert int_prop_element is not None
1276 assert float_prop_element is not None
1277
1278 string_prop: model.Property = string_prop_element # type: ignore
1279 int_prop: model.Property = int_prop_element # type: ignore
1280 float_prop: model.Property = float_prop_element # type: ignore
1281
1282 assert string_prop.value == shared_sme_string.value
1283 assert int(int_prop.value) == int(shared_sme_int.value)
1284 assert float(float_prop.value) == float(shared_sme_float.value)
1285
1286def test_036_get_submodel_by_id_metadata(wrapper: SdkWrapper, shared_sm: model.Submodel):
1287 sm_id = SM_ID
1288
1289 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1290 sm_id = encoder.encode_base_64(SM_ID)
1291
1292 metadata = wrapper.get_submodel_by_id_metadata(sm_id)
1293 assert metadata is not None
1294
1295 submodel = wrapper.get_submodel_by_id(sm_id)
1296 assert submodel is not None
1297
1298 assert metadata.get("id", "") == submodel.id
1299 assert metadata.get("idShort", "") == submodel.id_short
1300 assert metadata.get("description", {})[0].get("text", "") == submodel.description.get("en", "")
1301 assert metadata.get("displayName", {})[0].get("text", "") == submodel.display_name.get("en", "")
1302 assert "submodelElements" not in metadata
1303
1305 shell_id = SHELL_ID
1306
1307 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1308 shell_id = encoder.encode_base_64(SHELL_ID)
1309
1310 result = wrapper.delete_asset_administration_shell_by_id(shell_id)
1311
1312 assert result
1313
1314 shells = wrapper.get_all_asset_administration_shells()
1315 assert shells is not None
1316 assert len(shells.results) == 0
1317
1318def test_099a_delete_submodel_by_id(wrapper: SdkWrapper):
1319 sm_id = SM_ID
1320
1321 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1322 sm_id = encoder.encode_base_64(SM_ID)
1323
1324 result = wrapper.delete_submodel_by_id(sm_id)
1325
1326 assert result
1327
1328 submodels = wrapper.get_all_submodels()
1329 assert submodels is not None
1330 assert len(submodels.results) == 1
1331
1332def test_099b_delete_submodel_by_id(wrapper: SdkWrapper):
1333 sm_id = SM_ID
1334
1335 if wrapper.get_encoded_ids() == IdEncoding.encoded:
1336 sm_id = encoder.encode_base_64(AIMC_SM_ID)
1337
1338 result = wrapper.delete_submodel_by_id(sm_id)
1339
1340 assert result
1341
1342 submodels = wrapper.get_all_submodels()
1343 assert submodels is not None
1344 assert len(submodels.results) == 0
test_018a_patch_submodel_element_by_path_value_only_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_string)
test_031_delete_submodel_reference_by_id_aas_repository(SdkWrapper wrapper)
test_001a_connect(SdkWrapper wrapper)
test_000b_create_wrapper_by_dict(SdkWrapper wrapper)
test_032_put_submodel_element_by_path_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_string)
test_004a_get_asset_administration_shell_by_id(SdkWrapper wrapper, model.AssetAdministrationShell shared_aas)
test_023_put_file_content_by_path_submodel_repo(SdkWrapper wrapper)
test_025_get_thumbnail_aas_repository(SdkWrapper wrapper)
test_026_put_thumbnail_aas_repository(SdkWrapper wrapper)
test_018b_patch_submodel_element_by_path_value_only_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_bool)
test_033a_get_submodel_element_by_path_value_only_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_string)
test_017d_get_submodel_element_by_path_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_float)
test_016c_post_submodel_element_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_int)
test_005b_put_asset_administration_shell_by_id(SdkWrapper wrapper, model.AssetAdministrationShell shared_aas)
test_000a_create_wrapper_by_url(SdkWrapper wrapper)
test_005a_put_asset_administration_shell_by_id(SdkWrapper wrapper, model.AssetAdministrationShell shared_aas)
test_017c_get_submodel_element_by_path_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_int)
test_027_get_thumbnail_aas_repository(SdkWrapper wrapper)
test_021_post_file_by_path_submodel_repo(SdkWrapper wrapper)
test_020a_encoded_ids(SdkWrapper wrapper)
test_015_get_all_submodel_elements_submodel_repository(SdkWrapper wrapper)
test_018d_patch_submodel_element_by_path_value_only_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_float)
test_017a_get_submodel_element_by_path_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_string)
test_020b_encoded_ids(SdkWrapper wrapper)
test_002_get_all_asset_administration_shells(SdkWrapper wrapper)
test_019b_post_submodel_element_by_path_submodel_repo(SdkWrapper wrapper)
test_036_get_submodel_by_id_metadata(SdkWrapper wrapper, model.Submodel shared_sm)
test_007_get_submodel_by_id_aas_repository(SdkWrapper wrapper, model.Submodel shared_sm)
test_011b_get_submodel_by_id(SdkWrapper wrapper)
test_014_put_submodels_by_id(SdkWrapper wrapper, model.Submodel shared_sm)
test_016d_post_submodel_element_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_float)
model.AssetAdministrationShell shared_aas(model.Submodel shared_sm)
test_033c_get_submodel_element_by_path_value_only_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_float)
test_034_get_submodel_by_id_value_only(SdkWrapper wrapper, model.Submodel shared_sm)
model.Property shared_sme_int()
test_016b_post_submodel_element_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_bool)
model.Property shared_sme_float()
test_099a_delete_submodel_by_id(SdkWrapper wrapper)
test_035_patch_submodel_by_id_value_only(SdkWrapper wrapper, model.Submodel shared_sm, model.Property shared_sme_string, model.Property shared_sme_int, model.Property shared_sme_float)
model.Submodel shared_sm()
test_009a_post_submodel(SdkWrapper wrapper, model.Submodel shared_sm)
model.Property shared_sme_string()
test_029_get_all_submodel_references_aas_repository(SdkWrapper wrapper)
test_018c_patch_submodel_element_by_path_value_only_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_int)
test_011a_get_submodel_by_id(SdkWrapper wrapper, model.Submodel shared_sm)
test_024_delete_file_content_by_path_submodel_repo(SdkWrapper wrapper)
test_010_get_submodel_by_id_aas_repository(SdkWrapper wrapper, model.Submodel shared_sm)
test_099b_delete_submodel_by_id(SdkWrapper wrapper)
test_004b_get_asset_administration_shell_by_id(SdkWrapper wrapper)
test_001b_delete_all_asset_administration_shells(SdkWrapper wrapper)
test_006_get_asset_administration_shell_by_id_reference_aas_repository(SdkWrapper wrapper)
test_017b_get_submodel_element_by_path_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_bool)
model.Property shared_sme_bool()
test_033d_get_submodel_element_by_path_value_only_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_bool)
test_011d_get_submodel_by_id(SdkWrapper wrapper)
test_019a_post_submodel_element_by_path_submodel_repo(SdkWrapper wrapper)
test_008_get_all_submodels(SdkWrapper wrapper)
test_003_post_asset_administration_shell(SdkWrapper wrapper, model.AssetAdministrationShell shared_aas)
test_001c_delete_all_submodels(SdkWrapper wrapper)
test_033b_get_submodel_element_by_path_value_only_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_int)
test_000c_get_client(SdkWrapper wrapper)
test_009b_post_submodel(SdkWrapper wrapper)
test_011c_get_submodel_by_id(SdkWrapper wrapper)
test_013_put_submodel_by_id_aas_repository(SdkWrapper wrapper, model.Submodel shared_sm)
test_028_delete_thumbnail_aas_repository(SdkWrapper wrapper)
test_022_get_file_content_by_path_submodel_repo(SdkWrapper wrapper)
test_030_post_submodel_reference_aas_repository(SdkWrapper wrapper)
test_098_delete_asset_administration_shell_by_id(SdkWrapper wrapper)
test_012_patch_submodel_by_id(SdkWrapper wrapper, model.Submodel shared_sm)
SdkWrapper wrapper(request)
test_016a_post_submodel_element_submodel_repo(SdkWrapper wrapper, model.Property shared_sme_string)