diff --git a/.gitignore b/.gitignore
index 69a3c27..1fd581a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,5 +1,5 @@
# -------------------------------------------------------------------------
-# Copyright (c) 2020, PTC Inc. and/or all its affiliates. All rights reserved.
+# Copyright (c) PTC Inc. and/or all its affiliates. All rights reserved.
# See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
@@ -13,6 +13,7 @@ _notes
_new*
_test*
_build*
+_dev*
/docs
mkdocs.yml
diff --git a/README.md b/README.md
index 387a2b5..a0700e5 100644
--- a/README.md
+++ b/README.md
@@ -20,9 +20,9 @@ SDK allows for *GET*, *ADD*, *DELETE*, and *MODIFY* functions for the following
| **Connectivity**
*(Channel, Devices, Tags, Tag Groups)* | Y | Y |
| **IoT Gateway**
*(Agents, IoT Items)* | Y | Y |
| **Datalogger**
*(Log Groups, Items, Mapping, Triggers, Reset Mapping Service)* | Y | Y |
-| **Administration**
*(User Groups, Users, UA Endpoints)* | Y* | Y |
+| **Administration**
*(User Groups, Users, UA Endpoints, Local License Server)* | Y* | Y |
-Note (*) - UA Endpoints supported for Kepware Edge only
+Note (*) - UA Endpoints and Local License Server supported for Kepware Edge only
Driver specific features:
diff --git a/kepconfig/__init__.py b/kepconfig/__init__.py
index 7af9db2..73971e3 100644
--- a/kepconfig/__init__.py
+++ b/kepconfig/__init__.py
@@ -1,10 +1,10 @@
# -------------------------------------------------------------------------
-# Copyright (c) 2020, PTC Inc. and/or all its affiliates. All rights reserved.
+# Copyright (c) PTC Inc. and/or all its affiliates. All rights reserved.
# See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
-__version__ = "1.0.0"
+__version__ = "1.0.1"
__path__ = __import__("pkgutil").extend_path(__path__, __name__)
from . import connection, error
diff --git a/kepconfig/admin/__init__.py b/kepconfig/admin/__init__.py
index 844bedb..42079d3 100644
--- a/kepconfig/admin/__init__.py
+++ b/kepconfig/admin/__init__.py
@@ -1,8 +1,8 @@
# -------------------------------------------------------------------------
-# Copyright (c) 2020, PTC Inc. and/or all its affiliates. All rights reserved.
+# Copyright (c) PTC Inc. and/or all its affiliates. All rights reserved.
# See License.txt in the project root for
# license information.
# --------------------------------------------------------------------------
__path__ = __import__("pkgutil").extend_path(__path__, __name__)
-from . import users, user_groups,ua_server
\ No newline at end of file
+from . import users, user_groups,ua_server, lls
\ No newline at end of file
diff --git a/kepconfig/admin/lls.py b/kepconfig/admin/lls.py
new file mode 100644
index 0000000..10750d9
--- /dev/null
+++ b/kepconfig/admin/lls.py
@@ -0,0 +1,133 @@
+# -------------------------------------------------------------------------
+# Copyright (c), PTC Inc. and/or all its affiliates. All rights reserved.
+# See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+
+r"""`lls` exposes an API to allow modifications to Local License Server parameters in
+the Kepware Administration through the Kepware Configuration API
+"""
+from typing import Union
+from ..error import KepHTTPError, KepError
+
+
+LLS_ROOT = '/admin'
+LICENSING_SERVER_PORT = "libadminsettings.LICENSING_SERVER_PORT"
+LICENSING_SERVER_NAME = "libadminsettings.LICENSING_SERVER_NAME"
+LICENSING_SERVER_ENABLE = "libadminsettings.LICENSING_SERVER_ENABLE"
+LICENSING_CHECK_PERIOD_MINS = "libadminsettings.LICENSING_CHECK_PERIOD_MINS"
+LICENSING_SERVER_SSL_PORT = "libadminsettings.LICENSING_SERVER_SSL_PORT"
+LICENSING_SERVER_ALLOW_INSECURE_COMMS = "libadminsettings.LICENSING_SERVER_ALLOW_INSECURE_COMMS"
+LICENSING_SERVER_ALLOW_SELF_SIGNED_CERTS = "libadminsettings.LICENSING_SERVER_ALLOW_SELF_SIGNED_CERTS"
+LICENSING_CLIENT_ALIAS = "libadminsettings.LICENSING_CLIENT_ALIAS"
+
+class lls_config:
+ '''A class to represent a admin properties for the Local License Server connection from an instance of Kepware.
+ This object is used to easily manage the LLS parameters for a Kepware instance.
+
+ Properties:
+
+ "server_name" - Host name or IP address of the LLS server
+ "server_port" - HTTP/non-SSL port to target for the LLS server
+ "check_period" - Period that Kepware checks licensing status
+ "server_port_SSL" - HTTPS/SSL port to target for the LLS server
+ "allow_insecure_comms" - When True, use HTTP/non-SSL connection to LLS
+ "allow_self_signed_certs" - Allow for self signed certificates to be used during HTTPS/SSL connections to the LLS
+ "instance_alias_name" - Alias name for LLS to use as reference to this Kepware instance
+ '''
+
+ def __init__(self, config = {}):
+ self.server_name = config[LICENSING_SERVER_NAME] if LICENSING_SERVER_NAME in config else ''
+ self.server_port = config[LICENSING_SERVER_PORT] if LICENSING_SERVER_PORT in config else 7070
+ self.check_period = config[LICENSING_CHECK_PERIOD_MINS] if LICENSING_CHECK_PERIOD_MINS in config else 5
+ self.server_port_SSL = config[LICENSING_SERVER_SSL_PORT] if LICENSING_SERVER_SSL_PORT in config else 1883
+ self.allow_insecure_comms = config[LICENSING_SERVER_ALLOW_INSECURE_COMMS] if LICENSING_SERVER_ALLOW_INSECURE_COMMS in config else False
+ self.allow_self_signed_certs = config[LICENSING_SERVER_ALLOW_SELF_SIGNED_CERTS] if LICENSING_SERVER_ALLOW_SELF_SIGNED_CERTS in config else False
+ self.instance_alias_name = config[LICENSING_CLIENT_ALIAS] if LICENSING_CLIENT_ALIAS in config else ''
+
+ def _get_dict(self):
+ return {
+ LICENSING_SERVER_PORT: self.server_port,
+ LICENSING_SERVER_NAME: self.server_name,
+ LICENSING_CHECK_PERIOD_MINS: self.check_period,
+ LICENSING_SERVER_SSL_PORT: self.server_port_SSL,
+ LICENSING_SERVER_ALLOW_INSECURE_COMMS: self.allow_insecure_comms,
+ LICENSING_SERVER_ALLOW_SELF_SIGNED_CERTS: self.allow_self_signed_certs,
+ LICENSING_CLIENT_ALIAS: self.instance_alias_name
+ }
+
+ def __str__(self) -> str:
+ return "{}".format(self._get_dict())
+
+def get_lls_config(server) -> lls_config:
+ '''Returns the properties of the Local License server properties. Returned object is lls_config class object.
+
+ INPUTS:
+ "server" - instance of the "server" class
+
+ RETURNS:
+ lls_config - class object with lls configuration
+
+ EXCEPTIONS:
+ KepHTTPError - If urllib provides an HTTPError
+ KepURLError - If urllib provides an URLError
+ '''
+
+ r = server._config_get(server.url + LLS_ROOT)
+ return lls_config(r.payload)
+
+def update_lls_config(server, config: lls_config) -> bool:
+ '''Updates the Local License Server admin properties for Kepware.
+
+ INPUTS:
+ "server" - instance of the "server" class
+ "config" - lls_config class object
+
+ RETURNS:
+ True - If a "HTTP 200 - OK" is received from Kepware
+
+ EXCEPTIONS:
+ KepHTTPError - If urllib provides an HTTPError
+ KepURLError - If urllib provides an URLError
+ '''
+
+ DATA = config._get_dict()
+ r = server._config_update(server.url + LLS_ROOT, DATA)
+ if r.code == 200: return True
+ else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
+
+def enable_lls(server) -> bool:
+ '''Enables the Local License Server connection for Kepware.
+
+ INPUTS:
+ "server" - instance of the "server" class
+
+ RETURNS:
+ True - If a "HTTP 200 - OK" is received from Kepware
+
+ EXCEPTIONS:
+ KepHTTPError - If urllib provides an HTTPError
+ KepURLError - If urllib provides an URLError
+ '''
+
+ r = server._config_update(server.url + LLS_ROOT, {LICENSING_SERVER_ENABLE: True})
+ if r.code == 200: return True
+ else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
+
+def disable_lls(server) -> bool:
+ '''Disables the Local License Server connection for Kepware.
+
+ INPUTS:
+ "server" - instance of the "server" class
+
+ RETURNS:
+ True - If a "HTTP 200 - OK" is received from Kepware
+
+ EXCEPTIONS:
+ KepHTTPError - If urllib provides an HTTPError
+ KepURLError - If urllib provides an URLError
+ '''
+
+ r = server._config_update(server.url + LLS_ROOT, {LICENSING_SERVER_ENABLE: False})
+ if r.code == 200: return True
+ else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
diff --git a/kepconfig/connection.py b/kepconfig/connection.py
index c759f8c..955c55c 100644
--- a/kepconfig/connection.py
+++ b/kepconfig/connection.py
@@ -37,6 +37,25 @@ def __init__(self, code = '', message = '', href = ''):
def __str__(self):
return '{"code": %s, "message": %s, "href": %s}' % (self.code, self.message, self.href)
+class KepServiceStatus:
+ '''A class to represent a status object when checking on a "service" API job state in Kepware. This is
+ used to return the status of a "service" job
+
+ Properties:
+
+ "complete" - Boolean of service job completion status
+ "status" - Status code of job
+ "message" - Error message if service job fails
+
+ '''
+ def __init__(self, complete = '', status = '', message = ''):
+ self.status = status
+ self.message = message
+ self.complete = complete
+
+ def __str__(self):
+ return '{"complete": %s, "status": %s, "message": %s}' % (self.complete, self.status, self.message)
+
class _HttpDataAbstract:
def __init__(self):
self.payload = ''
@@ -62,9 +81,12 @@ class server:
Methods:
- "reinitialize()" - Reinitialize the Kepware server
+ "reinitialize()" - reinitialize the Kepware server
"get_trans_log()" - retrieve the Configuration API transaction logs
"get_event_log()" - retrieve the Kepware Event Log
+ "get_project_properties()" - retrieve the Kepware Project Properties
+ "modify_project_properties()" - modify the Kepware Project Properties
+ "service_status()" - retrive service job status
'''
__root_url = '/config'
__version_url = '/v1'
@@ -213,6 +235,30 @@ def modify_project_properties(self, DATA, force = False) -> bool:
r = self._config_update(self.url + '/project', prop_data)
if r.code == 200: return True
else: raise KepHTTPError(r.url, r.code, r.msg, r.hdrs, r.payload)
+
+ def service_status(self, resp: KepServiceResponse):
+ '''Returns the status of a service job. Used to verify if a service call
+ has completed or not.
+
+ INPUT:
+ "resp" - KepServiceResponse instance with job information
+
+ RETURNS:
+ KepServiceStatus instance with job status
+
+ EXCEPTIONS:
+
+ KepHTTPError - If urllib provides an HTTPError
+ KepURLError - If urllib provides an URLError
+ '''
+ # need to remove part of job href
+ loc = resp.href.find(self.__root_url + self.__version_url)
+ job_url = resp.href[loc + len(self.__root_url + self.__version_url):]
+
+ r = self._config_get(self.url + job_url)
+ job = KepServiceStatus(r.payload['servermain.JOB_COMPLETE'],r.payload['servermain.JOB_STATUS'], r.payload['servermain.JOB_STATUS_MSG'])
+ return job
+
#Function used to Add an object to Kepware (HTTP POST)
def _config_add(self, url, DATA):
@@ -279,7 +325,7 @@ def _force_update_check(self, force, DATA):
pass
return DATA
# General service call handler
- def _kep_service_execute(self, url, TTL):
+ def _kep_service_execute(self, url, TTL = None):
try:
if TTL != None:
TTL = {"servermain.JOB_TIME_TO_LIVE_SECONDS": TTL}
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..07de284
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,3 @@
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta"
\ No newline at end of file
diff --git a/release.py b/release.py
index 523b754..96d16ab 100644
--- a/release.py
+++ b/release.py
@@ -2,7 +2,7 @@
import os
file_location = "kepconfig/__init__.py"
-version_re_string = "([0-9]+)\.([0-9]+)\.([0-9]+$|[0-9]+[ab][0-9]+)"
+version_re_string = "([0-9]+)\.([0-9]+)\.([0-9]+(?:[ab][0-9])?)"
version_search = re.compile(r'__version__ = "'+ version_re_string + '"')
version_check = re.compile(version_re_string)
@@ -12,7 +12,7 @@ def bump_version():
m = version_search.search(s)
v1, v2, v3 = m.groups()
oldv = "{0}.{1}.{2}".format(v1, v2, v3)
- ans = input("Current version of kepconfig is: {0} \nUpdate new version to? (ctrl-c to exit): ".format(oldv))
+ ans = input("Current version of kepconfig is: {} \nUpdate new version to? (ctrl-c to exit): ".format(oldv))
if ans:
m = version_check.search(ans)
if (m==None):
@@ -22,7 +22,7 @@ def bump_version():
else:
print("Please enter updated version number. Exiting...")
exit()
- print("\n"+ "Updating " + file_location + " version to {0}.".format(newv))
+ print("\n"+ "Updating " + file_location + " version to {}.".format(newv))
s = s.replace(oldv, newv)
with open(file_location, "w") as f:
f.write(s)
@@ -36,22 +36,24 @@ def release():
os.system("git add " + file_location)
os.system('git commit -m \"{} Release\"'.format(v))
os.system("git tag {0}".format(v))
- ans = input("Change committed, push to server? (y/n)")
+ ans = input("Change committed, push to server? (Y/n)")
if ans.lower() in ("y", "yes"):
# os.system("git push")
# os.system("git push --tags")
os.system("git push --follow-tags")
- ans = input("upload to pip?(Y/n)")
- if ans.lower() in ("y", "yes"):
- # os.system("rm -rf dist/*") #Linux
- os.system("RMDIR /S /Q dist") #Windows
- os.system("python setup.py sdist bdist_wheel")
+ ans = input("Build dist packages?(Y/n)")
+ if ans.lower() in ("y", "yes"):
+ # os.system("rm -rf dist/*") #Linux
+ os.system("RMDIR /S /Q dist") #Windows
+ os.system("python -m build")
+ ans = input("upload to pip?(Y/n)")
+ if ans.lower() in ("y", "yes"):
- # Test PyPi Server
- os.system("twine upload --repository testpypi dist/*")
+ # Test PyPi Server
+ os.system("twine upload --repository testpypi dist/*")
- #Production PyPi Server
- # os.system("twine upload dist/*")
+ #Production PyPi Server
+ # os.system("twine upload dist/*")
if __name__ == "__main__":
diff --git a/samples/kepware lls config example.py b/samples/kepware lls config example.py
new file mode 100644
index 0000000..f2ead67
--- /dev/null
+++ b/samples/kepware lls config example.py
@@ -0,0 +1,77 @@
+# -------------------------------------------------------------------------
+# Copyright (c) PTC Inc. and/or all its affiliates. All rights reserved.
+# See License.txt in the project root for
+# license information.
+# --------------------------------------------------------------------------
+
+# Kepware LLS Config Example - Simple example on how to manage properties of a Kepware instance
+# to connect to a Local License Server through the Kepware Configuration API
+
+from kepconfig import connection, error
+from kepconfig.admin import lls
+
+def ErrorHandler(err):
+ # Generic Handler for exception errors
+ if err.__class__ is error.KepError:
+ print(err.msg)
+ elif err.__class__ is error.KepHTTPError:
+ print(err.code)
+ print(err.msg)
+ print(err.url)
+ print(err.hdrs)
+ print(err.payload)
+ elif err.__class__ is error.KepURLError:
+ print(err.url)
+ print(err.reason)
+ else:
+ print('Different Exception Received: {}'.format(err))
+
+# This creates a server reference that is used to target all modifications of
+# the Kepware configuration
+server = connection.server(host = '127.0.0.1', port = 57412, user = 'Administrator', pw = '')
+
+
+
+# Retreive the LLS properties from a Kepware instance and return a lls_config class object
+try:
+ LLSconfig = lls.get_lls_config(server)
+ print("{} - {}".format("Get Local License Server parameters for Kepware instance",LLSconfig))
+except Exception as err:
+ ErrorHandler(err)
+
+
+# Create a lls_config class object from Dict parameters
+JSON_lls_config = {
+ "libadminsettings.LICENSING_SERVER_PORT": 80,
+ "libadminsettings.LICENSING_SERVER_NAME": "test_host",
+ "libadminsettings.LICENSING_CHECK_PERIOD_MINS": 20,
+ "libadminsettings.LICENSING_SERVER_SSL_PORT": 7777,
+ "libadminsettings.LICENSING_SERVER_ALLOW_INSECURE_COMMS": True,
+ "libadminsettings.LICENSING_SERVER_ALLOW_SELF_SIGNED_CERTS": True,
+ "libadminsettings.LICENSING_CLIENT_ALIAS": "TestAliasName"
+}
+
+try:
+ print("{} - {}".format("Create Local License Server parameters object from Dict", lls.lls_config(JSON_lls_config)))
+except Exception as err:
+ ErrorHandler(err)
+
+
+# Update lls_config object with new values and update the Kepware instance with new parameters
+LLSconfig.server_name = 'HOSTNAME'
+try:
+ print("{} - {}".format("Update Local License Server parameters for Kepware instance",lls.update_lls_config(server, LLSconfig)))
+except Exception as err:
+ ErrorHandler(err)
+
+# Enable the LLS connection for the Kepware instance
+try:
+ print("{} - {}".format("Enable Local License Server connection for Kepware instance",lls.enable_lls(server)))
+except Exception as err:
+ ErrorHandler(err)
+
+# Disable the LLS connection for the Kepware instance
+try:
+ print("{} - {}".format("Disable Local License Server connection for Kepware instance",lls.disable_lls(server)))
+except Exception as err:
+ ErrorHandler(err)
\ No newline at end of file
diff --git a/samples/services and logs example.py b/samples/services and logs example.py
index aacdd01..9158e99 100644
--- a/samples/services and logs example.py
+++ b/samples/services and logs example.py
@@ -16,6 +16,8 @@
ch_name = 'ControlLogix_Channel'
dev_name = 'Device1'
device_IP = '192.168.1.100'
+# Service Response Global
+r = {}
def ErrorHandler(err):
# Generic Handler for exception errors
@@ -40,7 +42,15 @@ def ErrorHandler(err):
# This Reinitializes Kepware's Server Runtime process, similar to manually reinitializing
# using the Configuration UI or the Administrator tool.
try:
- print('{} - {}'.format("Execute Reinitialize Service", server.reinitialize()))
+ r = server.reinitialize()
+ print('{} - {}'.format("Execute Reinitialize Service", r))
+except Exception as err:
+ ErrorHandler(err)
+
+# This reads the Reinitialize Service Job Status using the returned KepServiceResponse
+
+try:
+ print('{} - {}'.format("Read Reinitialize Service Status", server.service_status(r)))
except Exception as err:
ErrorHandler(err)
@@ -67,9 +77,17 @@ def ErrorHandler(err):
# Execute the "TagGeneration" service available in the Kepware Configuration API
try:
- print("{} - {}".format("Executing ATG for Controllogix Device", device.auto_tag_gen(server, '{}.{}'.format(ch_name, dev_name))))
+ r = device.auto_tag_gen(server, '{}.{}'.format(ch_name, dev_name))
+ print("{} - {}".format("Executing ATG for Controllogix Device", r))
except Exception as err:
ErrorHandler(err)
+
+# This reads the TagGeneration Service Job Status using the returned KepServiceResponse
+try:
+ print('{} - {}'.format("Read Service Status", server.service_status(r)))
+except Exception as err:
+ ErrorHandler(err)
+
# Get Event Log from Kepware instance.
# Time parameters need to be UTC values.
try:
diff --git a/setup.cfg b/setup.cfg
deleted file mode 100644
index 98125b3..0000000
--- a/setup.cfg
+++ /dev/null
@@ -1,21 +0,0 @@
-# -------------------------------------------------------------------------
-# Copyright (c) 2020, PTC Inc. and/or all its affiliates. All rights reserved.
-# See License.txt in the project root for
-# license information.
-# --------------------------------------------------------------------------
-
-[metadata]
-# This includes the license file(s) in the wheel.
-# https://wheel.readthedocs.io/en/stable/user_guide.html#including-license-files-in-the-generated-wheel-file
-
-# license_files = LICENSE
-
-[bdist_wheel]
-# This flag says to generate wheels that support both Python 2 and Python
-# 3. If your code will not run unchanged on both Python 2 and 3, you will
-# need to generate separate wheels for each Python version that you
-# support. Removing this line (or setting universal to 0) will prevent
-# bdist_wheel from trying to make a universal wheel. For more see:
-# https://packaging.python.org/guides/distributing-packages-using-setuptools/#wheels
-
-universal=0
\ No newline at end of file
diff --git a/tests/admin_test.py b/tests/admin_test.py
index 7fb8aa2..1556e8a 100644
--- a/tests/admin_test.py
+++ b/tests/admin_test.py
@@ -45,6 +45,16 @@
"libadminsettings.UACONFIGMANAGER_ENDPOINT_PORT": 49333
}
+default_lls_config = {
+ "libadminsettings.LICENSING_SERVER_PORT": 7070,
+ "libadminsettings.LICENSING_SERVER_NAME": "",
+ "libadminsettings.LICENSING_CHECK_PERIOD_MINS": 5,
+ "libadminsettings.LICENSING_SERVER_SSL_PORT": 1443,
+ "libadminsettings.LICENSING_SERVER_ALLOW_INSECURE_COMMS": False,
+ "libadminsettings.LICENSING_SERVER_ALLOW_SELF_SIGNED_CERTS": False,
+ "libadminsettings.LICENSING_CLIENT_ALIAS": ""
+}
+
def HTTPErrorHandler(err):
if err.__class__ is KepHTTPError:
print(err.code)
@@ -74,6 +84,9 @@ def complete(server):
admin.user_groups.del_user_group(server,ug['common.ALLTYPES_NAME'])
except Exception as err:
pass
+
+ admin.lls.update_lls_config(server,admin.lls.lls_config(default_lls_config))
+
@pytest.fixture(scope="module")
def server(kepware_server):
@@ -86,9 +99,12 @@ def server(kepware_server):
yield server
complete(server)
-
def test_uaserver(server):
-
+ try:
+ server._config_get(server.url + admin.ua_server._create_url())
+ except Exception as err:
+ pytest.skip("UA Endpoints not configurable.")
+
assert admin.ua_server.add_endpoint(server,uaendpoint1)
assert admin.ua_server.del_endpoint(server,uaendpoint1['common.ALLTYPES_NAME'])
@@ -200,3 +216,24 @@ def test_users(server):
assert admin.users.del_user(server, user1['common.ALLTYPES_NAME'])
assert admin.users.del_user(server, user2['common.ALLTYPES_NAME'])
+
+def test_LLS(server):
+
+ assert type(admin.lls.get_lls_config(server)) == admin.lls.lls_config
+
+ lls_config = {"libadminsettings.LICENSING_SERVER_PORT": 80,
+ "libadminsettings.LICENSING_SERVER_NAME": "test_host",
+ "libadminsettings.LICENSING_CHECK_PERIOD_MINS": 20,
+ "libadminsettings.LICENSING_SERVER_SSL_PORT": 7777,
+ "libadminsettings.LICENSING_SERVER_ALLOW_INSECURE_COMMS": True,
+ "libadminsettings.LICENSING_SERVER_ALLOW_SELF_SIGNED_CERTS": True,
+ "libadminsettings.LICENSING_CLIENT_ALIAS": "Dumb"}
+
+ r = admin.lls.lls_config(lls_config)
+ assert type(r) == admin.lls.lls_config
+
+ assert admin.lls.update_lls_config(server, r)
+
+ assert admin.lls.enable_lls(server)
+
+ assert admin.lls.disable_lls(server)
\ No newline at end of file
diff --git a/tests/datalogger_test.py b/tests/datalogger_test.py
index 63b2fb5..8c844af 100644
--- a/tests/datalogger_test.py
+++ b/tests/datalogger_test.py
@@ -154,9 +154,9 @@ def HTTPErrorHandler(err):
else:
print('Different Exception Received: {}'.format(err))
-def initialize(server):
+def initialize(server: kepconfig.connection.server):
try:
- server._config_get(server.url +"/project/_datalogger")
+ server._config_get(server.url + datalogger.log_group._create_url())
except Exception as err:
pytest.skip("DataLogger plug-in is not installed", allow_module_level=True)
diff --git a/tests/egd_test.py b/tests/egd_test.py
index c52f144..850ec0a 100644
--- a/tests/egd_test.py
+++ b/tests/egd_test.py
@@ -117,7 +117,7 @@ def HTTPErrorHandler(err):
}]
}
-def initialize(server):
+def initialize(server: kepconfig.connection.server):
try:
server._config_get(server.url +"/doc/drivers/GE Ethernet Global Data/channels")
except Exception as err:
diff --git a/tests/iot_gateway_test.py b/tests/iot_gateway_test.py
index 788ee78..fb8310e 100644
--- a/tests/iot_gateway_test.py
+++ b/tests/iot_gateway_test.py
@@ -25,11 +25,10 @@
twx_agent_name = 'Thingworx'
iot_item_name ="System__Date"
-agent_list = [
- [mqtt_agent_name, kepconfig.iot_gateway.MQTT_CLIENT_AGENT],
- [rest_agent_name, kepconfig.iot_gateway.REST_CLIENT_AGENT],
- [rserver_agent_name, kepconfig.iot_gateway.REST_SERVER_AGENT]
- ]
+# used to test which agents are installed to test against.
+agent_list_avail = []
+
+agent_list = []
agent_data = {
"common.ALLTYPES_NAME": 'TempName',
@@ -71,8 +70,15 @@ def HTTPErrorHandler(err):
else:
print('Different Exception Received: {}'.format(err))
-def initialize(server):
- pass
+def initialize(server: kepconfig.connection.server):
+ try:
+ agent_list_avail = server._config_get(server.url +"/project/_iot_gateway?content=type_definition").payload
+ for agent in agent_list_avail['child_collections']:
+ if agent == 'mqtt_clients': agent_list.append([mqtt_agent_name, kepconfig.iot_gateway.MQTT_CLIENT_AGENT])
+ if agent == 'rest_clients': agent_list.append([rest_agent_name, kepconfig.iot_gateway.REST_CLIENT_AGENT])
+ if agent == 'rest_servers': agent_list.append([rserver_agent_name, kepconfig.iot_gateway.REST_SERVER_AGENT])
+ except Exception as err:
+ pytest.skip("IoT gateway plug-in is not installed", allow_module_level=True)
def complete(server):
# Delete all Agents
diff --git a/tests/server_connection_test.py b/tests/server_connection_test.py
index 0f9a5ab..a69fb1d 100644
--- a/tests/server_connection_test.py
+++ b/tests/server_connection_test.py
@@ -32,7 +32,7 @@ def complete(server):
pass
@pytest.fixture(scope="module")
-def server(kepware_server):
+def server(kepware_server: kepconfig.connection.server):
server = kepware_server
# Initialize any configuration before testing in module
@@ -42,7 +42,7 @@ def server(kepware_server):
yield server
complete(server)
-def test_connection_params(server):
+def test_connection_params(server: kepconfig.connection.server):
server.SSL_trust_all_certs = True
assert server.SSL_trust_all_certs == True
@@ -54,13 +54,17 @@ def test_connection_params(server):
server.SSL_ignore_hostname = False
assert server.SSL_ignore_hostname == False
-def test_reinitialize(server):
+def test_reinitialize_service_status(server: kepconfig.connection.server):
job = server.reinitialize()
assert type(job) == kepconfig.connection.KepServiceResponse
+ time.sleep(1)
+ status = server.service_status(job)
+ assert type(status) == kepconfig.connection.KepServiceStatus
job = server.reinitialize(60)
assert type(job) == kepconfig.connection.KepServiceResponse
-def test_project_props(server):
+
+def test_project_props(server: kepconfig.connection.server):
# Get Project Properties
assert type(server.get_project_properties()) == dict
@@ -79,7 +83,7 @@ def test_project_props(server):
}
assert server.modify_project_properties(project_prop, force = True)
-def test_event_log(server):
+def test_event_log(server: kepconfig.connection.server):
assert type(server.get_event_log(25, None, None)) == list
- assert type(server.get_event_log(25, datetime.datetime.fromisoformat('2019-11-03T23:35:23.000'), datetime.datetime.now())) == list
+ assert type(server.get_event_log(None, datetime.datetime.fromisoformat('2022-02-21T14:23:23.000'), datetime.datetime.utcnow())) == list