Skip to content

Commit 9824f86

Browse files
rlabbeptcrlabbeptc
committed
Added UDD profile support and testing
1 parent cac4511 commit 9824f86

4 files changed

Lines changed: 281 additions & 2 deletions

File tree

‎kepconfig/connectivity/__init__.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
# -------------------------------------------------------------------------
2-
# Copyright (c) 2020, PTC Inc. and/or all its affiliates. All rights reserved.
2+
# Copyright (c) PTC Inc. and/or all its affiliates. All rights reserved.
33
# See License.txt in the project root for
44
# license information.
55
# --------------------------------------------------------------------------
66

77

88
__path__ = __import__("pkgutil").extend_path(__path__, __name__)
9-
from . import channel, device, tag, egd
9+
from . import channel, device, tag, egd, udd

‎kepconfig/connectivity/udd/__init__.py‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
# -------------------------------------------------------------------------
2+
# Copyright (c) PTC Inc. and/or all its affiliates. All rights reserved.
3+
# See License.txt in the project root for
4+
# license information.
5+
# --------------------------------------------------------------------------
6+
7+
8+
__path__ = __import__("pkgutil").extend_path(__path__, __name__)
9+
from . import profile

‎kepconfig/connectivity/udd/profile.py‎

Lines changed: 160 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,160 @@
1+
# -------------------------------------------------------------------------
2+
# Copyright (c) PTC Inc. and/or all its affiliates. All rights reserved.
3+
# See License.txt in the project root for
4+
# license information.
5+
# --------------------------------------------------------------------------
6+
7+
8+
r"""`profile` exposes an API to allow modifications (add, delete, modify) to
9+
profile objects for the UDD Profile Library plug-in within the Kepware Configuration API
10+
"""
11+
12+
from kepconfig import connection
13+
from typing import Union
14+
15+
PROFILE_ROOT = '/project/_profile_library/profiles'
16+
17+
def add_profile(server: connection.server, DATA: Union[dict, list]) -> Union[bool, list]:
18+
'''Add a "profile" or a list of "profile" objects to the UDD Profile Library plug-in for Kepware.
19+
20+
Additionally it can be used to pass a list of exchanges and it's children to be added all at once.
21+
22+
INPUTS:
23+
24+
"server" - instance of the "server" class
25+
26+
"DATA" - properly JSON object (dict) of the exchange and it's children
27+
expected by Kepware Configuration API
28+
29+
RETURNS:
30+
31+
True - If a "HTTP 201 - Created" is received from Kepware
32+
33+
List - If a "HTTP 207 - Multi-Status" is received from Kepware with a list of dict error responses for all
34+
exchanges added that failed.
35+
36+
False - If a non-expected "2xx successful" code is returned
37+
38+
39+
EXCEPTIONS:
40+
41+
KepHTTPError - If urllib provides an HTTPError
42+
KepURLError - If urllib provides an URLError
43+
'''
44+
45+
r = server._config_add(f'{server.url}{PROFILE_ROOT}', DATA)
46+
if r.code == 201: return True
47+
elif r.code == 207:
48+
errors = []
49+
for item in r.payload:
50+
if item['code'] != 201:
51+
errors.append(item)
52+
return errors
53+
else: return False
54+
55+
def del_profile(server: connection.server, profile_name: str) -> bool:
56+
'''Delete a "profile" object in UDD Profile Library plug-in for Kepware.
57+
58+
INPUTS:
59+
60+
"server" - instance of the "server" class
61+
62+
"profile_name" - name of profile
63+
64+
RETURNS:
65+
66+
True - If a "HTTP 200 - OK" is received from Kepware
67+
68+
EXCEPTIONS:
69+
70+
KepHTTPError - If urllib provides an HTTPError
71+
KepURLError - If urllib provides an URLError
72+
'''
73+
74+
r = server._config_del(f'{server.url}{PROFILE_ROOT}/{profile_name}')
75+
if r.code == 200: return True
76+
else: return False
77+
78+
def modify_profile(server: connection.server, DATA: dict, profile_name: str = None, force: bool = False) -> bool:
79+
'''Modify a profile object and it's properties in Kepware. If a "profile_name" is not provided as an input,
80+
you need to identify the profile in the 'common.ALLTYPES_NAME' property field in the "DATA". It will
81+
assume that is the profile that is to be modified.
82+
83+
INPUTS:
84+
85+
"server" - instance of the "server" class
86+
87+
"DATA" - properly JSON object (dict) of the exchange properties to be modified.
88+
89+
"profile_name" (optional) - name of exchange to modify. Only needed if not existing in "DATA"
90+
91+
"force" (optional) - if True, will force the configuration update to the Kepware server
92+
93+
RETURNS:
94+
95+
True - If a "HTTP 200 - OK" is received from Kepware
96+
97+
EXCEPTIONS:
98+
99+
KepHTTPError - If urllib provides an HTTPError
100+
KepURLError - If urllib provides an URLError
101+
'''
102+
103+
profile_data = server._force_update_check(force, DATA)
104+
if profile_name == None:
105+
try:
106+
r = server._config_update(f"{server.url}{PROFILE_ROOT}/{profile_data['common.ALLTYPES_NAME']}", profile_data)
107+
if r.code == 200: return True
108+
else: return False
109+
except KeyError as err:
110+
print('Error: No profile identified in DATA | Key Error: {}'.format(err))
111+
return False
112+
# except Exception as e:
113+
# return 'Error: Error with {}: {}'.format(inspect.currentframe().f_code.co_name, str(e))
114+
else:
115+
r = server._config_update(f'{server.url}{PROFILE_ROOT}/{profile_name}', profile_data)
116+
if r.code == 200: return True
117+
else: return False
118+
119+
def get_profile(server: connection.server, profile_name: str = None) -> Union[dict, list]:
120+
'''Returns the properties of the profile object or a list of all profiles and their
121+
properties. Returned object is JSON.
122+
123+
INPUTS:
124+
125+
"server" - instance of the "server" class
126+
127+
"profile_name" - (optional) name of exchange. If not defined, get all profiles
128+
129+
RETURNS:
130+
131+
JSON - data for the exchange requested or a list of exchanges and their properties
132+
133+
EXCEPTIONS:
134+
135+
KepHTTPError - If urllib provides an HTTPError
136+
KepURLError - If urllib provides an URLError
137+
'''
138+
if profile_name == None:
139+
r = server._config_get(f'{server.url}{PROFILE_ROOT}')
140+
else:
141+
r = server._config_get(f'{server.url}{PROFILE_ROOT}/{profile_name}')
142+
return r.payload
143+
144+
def get_all_profiles(server: connection.server):
145+
'''Returns list of all profile objects and their properties. Returned object is JSON list.
146+
147+
INPUTS:
148+
149+
"server" - instance of the "server" class
150+
151+
RETURNS:
152+
153+
List - list of all profiles in the Profile Library
154+
155+
EXCEPTIONS:
156+
157+
KepHTTPError - If urllib provides an HTTPError
158+
KepURLError - If urllib provides an URLError
159+
'''
160+
return get_profile(server)

‎tests/udd_test.py‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
# -------------------------------------------------------------------------
2+
# Copyright (c) PTC Inc. All rights reserved.
3+
# See License.txt in the project root for
4+
# license information.
5+
# --------------------------------------------------------------------------
6+
7+
# UDD Test - Test to exersice all UDD Profile Library related features
8+
9+
import os, sys
10+
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
11+
import kepconfig
12+
from kepconfig import error, connectivity
13+
import kepconfig.connectivity.udd.profile as uddprofile
14+
import time
15+
import datetime
16+
import pytest
17+
18+
19+
# Channel and Device name to be used
20+
profile1 = {
21+
"common.ALLTYPES_NAME": "Profile1",
22+
"common.ALLTYPES_DESCRIPTION": "",
23+
"libudcommon.LIBUDCOMMON_PROFILE_JAVASCRIPT": "function onProfileLoad () {return {version: \"2.0\", mode: \"Client\"};}\nfunction onValidateTag (info) {}\nfunction onTagsRequest (info) {}\nfunction onData (info) {}\n"
24+
}
25+
profile2 = {
26+
"common.ALLTYPES_NAME": "Profile2",
27+
"common.ALLTYPES_DESCRIPTION": "",
28+
"libudcommon.LIBUDCOMMON_PROFILE_JAVASCRIPT": "function onProfileLoad () {return {version: \"2.0\", mode: \"Client\"};}\nfunction onValidateTag (info) {}\nfunction onTagsRequest (info) {}\nfunction onData (info) {}\n"
29+
}
30+
31+
def ErrorHandler(err):
32+
# Generic Handler for exception errors
33+
if err.__class__ is error.KepError:
34+
print(err.msg)
35+
elif err.__class__ is error.KepHTTPError:
36+
print(err.code)
37+
print(err.msg)
38+
print(err.url)
39+
print(err.hdrs)
40+
print(err.payload)
41+
elif err.__class__ is error.KepURLError:
42+
print(err.url)
43+
print(err.reason)
44+
else:
45+
print('Different Exception Received: {}'.format(err))
46+
47+
48+
def initialize(server: kepconfig.connection.server):
49+
if server_type == 'TKE': pytest.skip("UDD not configurable in {}.".format(server_type), allow_module_level=True)
50+
51+
try:
52+
server._config_get(server.url +"/doc/drivers/Universal Device/channels")
53+
except Exception as err:
54+
pytest.skip("UDD Driver is not installed", allow_module_level=True)
55+
56+
def complete(server):
57+
try:
58+
profiles_list = uddprofile.get_all_profiles(server)
59+
[uddprofile.del_profile(server, g['common.ALLTYPES_NAME']) for g in profiles_list]
60+
except Exception as err:
61+
ErrorHandler(err)
62+
63+
@pytest.fixture(scope="module")
64+
def server(kepware_server):
65+
server = kepware_server[0]
66+
global server_type
67+
server_type = kepware_server[1]
68+
69+
# Initialize any configuration before testing in module
70+
initialize(server)
71+
72+
# Everything below yield is run after module tests are completed
73+
yield server
74+
complete(server)
75+
76+
#
77+
# MAIN TEST SET
78+
#
79+
80+
81+
def test_profile_add(server: kepconfig.connection.server):
82+
# Add profile
83+
assert uddprofile.add_profile(server, profile1)
84+
85+
# Add multi profiles with one failure
86+
assert type(uddprofile.add_profile(server, [profile1, profile2])) == list
87+
88+
def test_profile_get(server: kepconfig.connection.server):
89+
# Get All Profiles
90+
assert type(uddprofile.get_all_profiles(server)) == list
91+
92+
# Get All Profiles - alternate
93+
assert type(uddprofile.get_profile(server)) == list
94+
95+
# Get one profile
96+
assert type(uddprofile.get_profile(server, profile1['common.ALLTYPES_NAME'])) == dict
97+
98+
99+
def test_profile_modify(server: kepconfig.connection.server):
100+
# Modify Profile
101+
profile1['common.ALLTYPES_DESCRIPTION'] = 'Test Change'
102+
assert uddprofile.modify_profile(server, profile1)
103+
104+
# Modify Profile with name parameter
105+
change = {'common.ALLTYPES_DESCRIPTION':'Test Change'}
106+
assert uddprofile.modify_profile(server, change, profile1['common.ALLTYPES_NAME'], force=True)
107+
108+
def test_profile_delete(server: kepconfig.connection.server):
109+
# Modify Profile
110+
assert uddprofile.del_profile(server, profile1['common.ALLTYPES_NAME'])

0 commit comments

Comments
 (0)