Skip to content

Use the solution from 38 #43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jacomago wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Use the solution from 38 #43

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
eb08e27
Make TLS certificate verification configurable.
smarsching Nov 12, 2025
cb46369
Simplify get_verify
jacomago Nov 19, 2025
e7c7b96
Use the verify_ssl option in tests
jacomago Mar 20, 2026
7506072
Match the pydoc
jacomago Mar 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,14 @@ username=MyUserName
password=MyPassword
```

If you are using an `https://` URL and want to disable certificate
verification, you can add:

```
verify_ssl=false
```


## Development

To install with dependancies for testing.
Expand Down
74 changes: 34 additions & 40 deletions channelfinder/ChannelFinderClient.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
from requests.exceptions import HTTPError
import urllib3
from copy import copy
from typing import Optional

try:
from json import JSONEncoder
Expand All @@ -29,23 +30,27 @@
__propertiesResource = "/resources/properties"
__tagsResource = "/resources/tags"

def __init__(self, BaseURL=None, username=None, password=None, verify_ssl=True):
def __init__(self, BaseURL=None, username=None, password=None, verify_ssl=None):

Check warning on line 33 in channelfinder/ChannelFinderClient.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Rename this parameter "BaseURL" to match the regular expression ^[_a-z][a-z0-9_]*$. 


See more on https://sonarcloud.io/project/issues?id=ChannelFinder_pyCFClient&issues=AZ0McyWq4WP0yRQjLj3z&open=AZ0McyWq4WP0yRQjLj3z&pullRequest=43
"""
Channel finder client object. It provides a connection object to perform the following operations:
- find: find all channels satisfying given searching criteria
- set: add channel into service
- update: update channel information
- delete: delete channel from service

If any of the arguments is ``None``, the respective value from the
configuration file is used.

:param BaseURL: the url of the channel finder service
:param username: user name authorized by channel finder service
:param password: password for the authorized user
:param verify_ssl: verify the peer TLS certificate
"""
try:
self.__baseURL = self.__getDefaultConfig("BaseURL", BaseURL)
self.__userName = self.__getDefaultConfig("username", username)
self.__password = self.__getDefaultConfig("password", password)
self.__verify_ssl = self.__getDefaultConfig("verify_ssl", verify_ssl)
self.__verify_ssl = self._get_boolean_config("verify_ssl", verify_ssl)
if self.__userName and self.__password:
self.__auth = auth.HTTPBasicAuth(self.__userName, self.__password)
else:
Expand All @@ -58,19 +63,34 @@
except Exception as e:
raise RuntimeError("Error creating ChannelFinderClient: " + str(e))

def __getDefaultConfig(self, key, ref):
def __getDefaultConfig(self, key, override):

Check warning on line 66 in channelfinder/ChannelFinderClient.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Rename method "__getDefaultConfig" to match the regular expression ^[a-z_][a-z0-9_]*$. 


See more on https://sonarcloud.io/project/issues?id=ChannelFinder_pyCFClient&issues=AZ0McyWq4WP0yRQjLj30&open=AZ0McyWq4WP0yRQjLj30&pullRequest=43
"""
Get default configuration for given name and section.

:param key: key word
:param ref: reference value
:return: result if key word is configured or ref is not None, otherwise None
:param override: override value
:return: ``override`` if not ``None``, else the configuration value
associated with ``key`` if present, otherwise ``None``.
"""
result = ref
if ref is None:
result = override
if override is None:
result = basecfg["DEFAULT"].get(key, None)
return result

def _get_boolean_config(self, key: str, override: Optional[bool]) -> bool:
"""
Get boolean configuration for given name and section.

:param key: key word
:param override: override value
:return: ``override`` if not ``None``, else the configuration value
associated with ``key`` if present, otherwise ``None``.
"""
result = override
if override is None:
result = basecfg["DEFAULT"].getboolean(key, None)
return bool(result)

def set(self, **kwds):
"""
method to allow various types of set operations on one or many channels, tags or properties
Expand Down Expand Up @@ -161,7 +181,6 @@
+ kwds["channel"]["name"],
data=JSONEncoder().encode(kwds["channel"]),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
)
r.raise_for_status()
Expand All @@ -170,7 +189,6 @@
self.__baseURL + self.__channelsResource,
data=JSONEncoder().encode(kwds["channels"]),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
)
r.raise_for_status()
Expand All @@ -179,7 +197,6 @@
self.__baseURL + self.__tagsResource + "/" + kwds["tag"]["name"],
data=JSONEncoder().encode(kwds["tag"]),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
)
r.raise_for_status()
Expand All @@ -189,7 +206,6 @@
self.__baseURL + self.__tagsResource,
data=data,
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
)
r.raise_for_status()
Expand All @@ -201,7 +217,6 @@
+ kwds["property"]["name"],
data=JSONEncoder().encode(kwds["property"]),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
)
r.raise_for_status()
Expand All @@ -212,7 +227,6 @@
self.__baseURL + self.__propertiesResource,
data=data,
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
)
r.raise_for_status()
Expand All @@ -238,7 +252,6 @@
self.__baseURL + self.__tagsResource + "/" + kwds["tag"]["name"],
data=JSONEncoder().encode(data),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
).raise_for_status()
elif "tag" in kwds and "channelNames" in kwds:
Expand All @@ -251,7 +264,6 @@
self.__baseURL + self.__tagsResource + "/" + kwds["tag"]["name"],
data=JSONEncoder().encode(data),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
).raise_for_status()
elif "property" in kwds and "channels" in kwds:
Expand All @@ -264,7 +276,6 @@
+ kwds["property"]["name"],
data=JSONEncoder().encode(data),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
).raise_for_status()
else:
Expand Down Expand Up @@ -365,7 +376,6 @@
url,
params=args,
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
)
try:
Expand All @@ -385,9 +395,7 @@
:return: Tag object if found, otherwise None
"""
url = self.__baseURL + self.__tagsResource + "/" + tagname
r = self.__session.get(
url, headers=copy(self.__jsonheader), verify=False, auth=self.__auth
)
r = self.__session.get(url, headers=copy(self.__jsonheader), auth=self.__auth)
try:
r.raise_for_status()
return r.json()
Expand All @@ -405,7 +413,7 @@
:return: Property object if found, otherwise None
"""
url = self.__baseURL + self.__propertiesResource + "/" + propertyname
r = self.__session.get(url, headers=copy(self.__jsonheader), verify=False)
r = self.__session.get(url, headers=copy(self.__jsonheader))
try:
r.raise_for_status()
return r.json()
Expand All @@ -422,7 +430,7 @@
:return: list of all the Tag objects present, otherwise None.
"""
url = self.__baseURL + self.__tagsResource
r = self.__session.get(url, headers=copy(self.__jsonheader), verify=False)
r = self.__session.get(url, headers=copy(self.__jsonheader))
try:
r.raise_for_status()
return r.json()
Expand All @@ -439,7 +447,7 @@
:return: list of the Property objects present, otherwise None
"""
url = self.__baseURL + self.__propertiesResource
r = self.__session.get(url, headers=copy(self.__jsonheader), verify=False)
r = self.__session.get(url, headers=copy(self.__jsonheader))
try:
r.raise_for_status()
return r.json()
Expand Down Expand Up @@ -504,12 +512,12 @@
+ kwds["channelName"].strip()
)
self.__session.delete(
url, headers=copy(self.__jsonheader), verify=False, auth=self.__auth
url, headers=copy(self.__jsonheader), auth=self.__auth
).raise_for_status()
elif "tagName" in kwds:
url = self.__baseURL + self.__tagsResource + "/" + kwds["tagName"].strip()
self.__session.delete(
url, verify=False, headers=copy(self.__jsonheader), auth=self.__auth
url, headers=copy(self.__jsonheader), auth=self.__auth
).raise_for_status()
elif "propertyName" in kwds:
url = (
Expand All @@ -519,7 +527,7 @@
+ kwds["propertyName"].strip()
)
self.__session.delete(
url, headers=copy(self.__jsonheader), verify=False, auth=self.__auth
url, headers=copy(self.__jsonheader), auth=self.__auth
).raise_for_status()
else:
raise RuntimeError(
Expand All @@ -545,7 +553,6 @@
+ "/"
+ kwds["channelName"].strip(),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
).raise_for_status()
elif "tag" in kwds and "channelNames" in kwds:
Expand All @@ -568,7 +575,6 @@
+ "/"
+ kwds["channelName"],
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
).raise_for_status()
elif "property" in kwds and "channelNames" in kwds:
Expand Down Expand Up @@ -674,7 +680,6 @@
self.__baseURL + self.__channelsResource + "/" + ch["name"],
data=JSONEncoder().encode(ch),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
)
r.raise_for_status()
Expand All @@ -684,7 +689,6 @@
self.__baseURL + self.__channelsResource,
data=JSONEncoder().encode(chs),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
)
r.raise_for_status()
Expand All @@ -694,7 +698,6 @@
self.__baseURL + self.__propertiesResource + "/" + property["name"],
data=JSONEncoder().encode(property),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
)
r.raise_for_status()
Expand All @@ -704,7 +707,6 @@
self.__baseURL + self.__tagsResource + "/" + tag["name"],
data=JSONEncoder().encode(tag),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
)
r.raise_for_status()
Expand All @@ -713,7 +715,6 @@
self.__baseURL + self.__tagsResource,
data=JSONEncoder().encode(kwds["tags"]),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
)
r.raise_for_status()
Expand Down Expand Up @@ -750,7 +751,6 @@
self.__baseURL + self.__tagsResource + "/" + tag["name"],
data=JSONEncoder().encode(tag),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
).raise_for_status()
elif "tag" in kwds and "channelNames" in kwds:
Expand All @@ -767,7 +767,6 @@
self.__baseURL + self.__tagsResource + "/" + tag["name"],
data=JSONEncoder().encode(tag),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
).raise_for_status()
elif "property" in kwds and "channelName" in kwds:
Expand All @@ -786,7 +785,6 @@
self.__baseURL + self.__propertiesResource + "/" + property["name"],
data=JSONEncoder().encode(property),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
).raise_for_status()
elif "property" in kwds and "channelNames" in kwds:
Expand All @@ -807,7 +805,6 @@
self.__baseURL + self.__propertiesResource + "/" + property["name"],
data=JSONEncoder().encode(property),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
).raise_for_status()
elif "originalChannelName" in kwds and "channel" in kwds:
Expand All @@ -817,7 +814,6 @@
self.__baseURL + self.__channelsResource + "/" + channelName,
data=JSONEncoder().encode(ch),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
).raise_for_status()
elif "originalPropertyName" in kwds and "property" in kwds:
Expand All @@ -827,7 +823,6 @@
self.__baseURL + self.__propertiesResource + "/" + propName,
data=JSONEncoder().encode(prop),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
).raise_for_status()
elif "originalTagName" in kwds and "tag" in kwds:
Expand All @@ -837,7 +832,6 @@
self.__baseURL + self.__tagsResource + "/" + tagName,
data=JSONEncoder().encode(tag),
headers=copy(self.__jsonheader),
verify=False,
auth=self.__auth,
).raise_for_status()
else:
Expand Down
1 change: 1 addition & 0 deletions test/_testConf.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ def __loadConfig():
"username": "admin",
"password": "adminPass",
"owner": "cf-update",
"verify_ssl": "False",
"channelOwner": "cf-channels",
"channelUsername": "admin",
"channelPassword": "adminPass",
Expand Down
4 changes: 0 additions & 4 deletions test/testCFPpropertyManager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,12 +6,8 @@

from _testConf import _testConf, ChannelFinderClientTestCase

import urllib3

from channelfinder.cfPropertyManager import CFPropertyManager

urllib3.disable_warnings()


class CFPropertyManagerTest(ChannelFinderClientTestCase):
def setUp(self):
Expand Down
4 changes: 0 additions & 4 deletions test/testCFUpdateIOC.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,6 @@

from _testConf import _testConf, ChannelFinderClientTestCase

import urllib3

urllib3.disable_warnings()


class UpdateIOCTest(ChannelFinderClientTestCase):
def setUp(self):
Expand Down
4 changes: 0 additions & 4 deletions test/testChannelFinderClient.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,10 +13,6 @@
from channelfinder.util import ChannelUtil
from _testConf import _testConf, ChannelFinderClientTestCase

import urllib3

urllib3.disable_warnings()


class ConnectionTest(ChannelFinderClientTestCase):
def testConnection(self):
Expand Down
Loading
Loading