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