Skip to content
GitLab
Commit 79230435 authored by Michael Thies's avatar Michael Thies
Browse files

Merge branch fix/couchdb_update_commit_things with refs/heads/master into... 

Merge branch fix/couchdb_update_commit_things with refs/heads/master into refs/merge-requests/64/train
parents 766aa5c7 bb1320b9
Pipeline #369540 passed with stage
in 1 minute and 12 seconds
Hide whitespace changes
Inline Side-by-side
README.md
View file @ 79230435
......@@ -40,6 +40,7 @@ PyI40AAS requires the following Python packages to be installed for production u
`setup.py` to be fetched automatically when installing with `pip`:
* `python-dateutil` (BSD 3-clause License)
* `lxml` (BSD 3-clause License, using `libxml2` under MIT License)
* `urllib3` (MIT License)
* `pyecma376-2` (Apache License v2.0)
Optional production usage dependencies:
......
aas/backend/couchdb.py
View file @ 79230435
......@@ -13,14 +13,12 @@ Todo: Add module docstring
"""
import threading
import weakref
from typing import List, Dict, Any, Optional, Iterator, Iterable, Union
import re
from typing import List, Dict, Any, Optional, Iterator, Iterable, Union, Tuple
import urllib.parse
import urllib.request
import urllib.error
import logging
import json
import http.client
import urllib3 # type: ignore
from . import backends
from aas.adapter.json import json_deserialization, json_serialization
......@@ -30,6 +28,9 @@ from aas import model
logger = logging.getLogger(__name__)
_http_pool_manager = urllib3.PoolManager()
class CouchDBBackend(backends.Backend):
"""
This Backend stores each Identifiable object as a single JSON document in the configured CouchDB database. Each
......@@ -39,18 +40,16 @@ class CouchDBBackend(backends.Backend):
"""
@classmethod
def update_object(cls,
updated_object: "Referable", # type: ignore
store_object: "Referable", # type: ignore
updated_object: model.Referable,
store_object: model.Referable,
relative_path: List[str]) -> None:
if not isinstance(store_object, model.Identifiable):
raise CouchDBSourceError("The given store_object is not Identifiable, therefore cannot be found "
"in the CouchDB")
url = CouchDBBackend._parse_source(store_object.source)
request = urllib.request.Request(url,
headers={'Accept': 'application/json'})
try:
data = CouchDBBackend.do_request(request)
data = CouchDBBackend.do_request(url)
except CouchDBServerError as e:
if e.code == 404:
raise KeyError("No Identifiable found in CouchDB at {}".format(url)) from e
......@@ -62,8 +61,8 @@ class CouchDBBackend(backends.Backend):
@classmethod
def commit_object(cls,
committed_object: "Referable", # type: ignore
store_object: "Referable", # type: ignore
committed_object: model.Referable,
store_object: model.Referable,
relative_path: List[str]) -> None:
if not isinstance(store_object, model.Identifiable):
raise CouchDBSourceError("The given store_object is not Identifiable, therefore cannot be found "
......@@ -75,13 +74,9 @@ class CouchDBBackend(backends.Backend):
data = json.dumps({'data': store_object, "_rev": get_couchdb_revision(url)},
cls=json_serialization.AASToJsonEncoder)
request = urllib.request.Request(
url,
headers={'Content-type': 'application/json'},
method='PUT',
data=data.encode())
try:
response = CouchDBBackend.do_request(request)
response = CouchDBBackend.do_request(
url, method='PUT', additional_headers={'Content-type': 'application/json'}, body=data.encode('utf-8'))
set_couchdb_revision(url, response["rev"])
except CouchDBServerError as e:
if e.code == 409:
......@@ -101,69 +96,82 @@ class CouchDBBackend(backends.Backend):
:return: URL to the document
:raises CouchDBBackendSourceError, if the source has the wrong format
"""
couchdb_s = re.match("couchdbs://", source) # Note: Works, since match only checks the beginning of the string
if couchdb_s:
if source.startswith("couchdbs://"):
url = source.replace("couchdbs://", "https://", 1)
elif source.startswith("couchdb://"):
url = source.replace("couchdb://", "http://", 1)
else:
couchdb_wo_s = re.match("couchdb://", source)
if couchdb_wo_s:
url = source.replace("couchdb://", "http://", 1)
else:
raise CouchDBSourceError("Source has wrong format. "
"Expected to start with {couchdb://, couchdbs://}, got {" + source + "}")
raise CouchDBSourceError("Source has wrong format. "
"Expected to start with {couchdb://, couchdbs://}, got {" + source + "}")
return url
@classmethod
def do_request(cls, request: urllib.request.Request) -> Dict[str, Any]:
def do_request(cls, url: str, method: str = "GET", additional_headers: Dict[str, str] = {},
body: Optional[bytes] = None) -> Dict[str, Any]:
"""
Perform an HTTP request to the CouchDBServer, parse the result and handle errors
:param request:
:return:
Perform an HTTP(S) request to the CouchDBServer, parse the result and handle errors
:param url: The HTTP or HTTPS URL to request
:param method: The HTTP method for the request
:param additional_headers: Additional headers to insert into the request. The default headers include
'connection: keep-alive', 'accept-encoding: ...', 'authorization: basic ...', 'Accept: ...'.
:param body: Request body for POST, PUT, and PATCH requests
:return: The parsed JSON data if the request `method` is other than 'HEAD' or the response headers for 'HEAD'
requests
"""
opener = urllib.request.build_opener(urllib.request.HTTPBasicAuthHandler(_credentials_store))
url_parts = urllib.parse.urlparse(url)
host = url_parts.scheme + url_parts.netloc
auth = _credentials_store.get(host)
headers = urllib3.make_headers(keep_alive=True, accept_encoding=True,
basic_auth="{}:{}".format(*auth) if auth else None)
headers['Accept'] = 'application/json'
headers.update(additional_headers)
try:
response = opener.open(request)
except urllib.error.HTTPError as e:
with e: # close the reponse (socket) when done
logger.debug("Request %s %s finished with HTTP status code %s.",
request.get_method(), request.full_url, e.code)
if e.headers.get('Content-type', None) != 'application/json':
raise CouchDBResponseError("Unexpected Content-type header {} of response from CouchDB server"
.format(e.headers.get('Content-type', None)))
if request.get_method() == 'HEAD':
raise CouchDBServerError(e.code, "", "", "HTTP {}") from e
try:
data = json.load(e)
except json.JSONDecodeError:
raise CouchDBResponseError("Could not parse error message of HTTP {}"
.format(e.code))
raise CouchDBServerError(e.code, data['error'], data['reason'],
"HTTP {}: {} (reason: {})".format(e.code, data['error'], data['reason']))\
from e
except urllib.error.URLError as e:
response = _http_pool_manager.request(method, url, headers=headers, body=body)
except (urllib3.exceptions.TimeoutError, urllib3.exceptions.SSLError, urllib3.exceptions.ProtocolError) as e:
raise CouchDBConnectionError("Error while connecting to the CouchDB server: {}".format(e)) from e
except urllib3.exceptions.HTTPError as e:
raise CouchDBResponseError("Error while connecting to the CouchDB server: {}".format(e)) from e
if not (200 <= response.status < 300):
logger.debug("Request %s %s finished with HTTP status code %s.",
method, url, response.status)
if response.headers.get('Content-type', None) != 'application/json':
raise CouchDBResponseError("Unexpected Content-type header {} of response from CouchDB server"
.format(response.headers.get('Content-type', None)))
if method == 'HEAD':
raise CouchDBServerError(response.status, "", "", "HTTP {}".format(response.status))
# Check response & parse data
assert (isinstance(response, http.client.HTTPResponse))
with response: # close the reponse (socket) when done
logger.debug("Request %s %s finished successfully.", request.get_method(), request.full_url)
if request.get_method() == 'HEAD':
return {}
if response.getheader('Content-type') != 'application/json':
raise CouchDBResponseError("Unexpected Content-type header")
try:
data = json.load(response, cls=json_deserialization.AASFromJsonDecoder)
except json.JSONDecodeError as e:
raise CouchDBResponseError("Could not parse CouchDB server response as JSON data.") from e
return data
data = json.loads(response.data.decode('utf-8'))
except json.JSONDecodeError:
raise CouchDBResponseError("Could not parse error message of HTTP {}"
.format(response.status))
raise CouchDBServerError(response.status, data['error'], data['reason'],
"HTTP {}: {} (reason: {})".format(response.status, data['error'], data['reason']))
# Check response & parse data
logger.debug("Request %s %s finished successfully.", method, url)
if method == 'HEAD':
return response.headers
if response.getheader('Content-type') != 'application/json':
raise CouchDBResponseError("Unexpected Content-type header")
try:
data = json.loads(response.data.decode('utf-8'), cls=json_deserialization.AASFromJsonDecoder)
except json.JSONDecodeError as e:
raise CouchDBResponseError("Could not parse CouchDB server response as JSON data.") from e
return data
backends.register_backend("couchdb", CouchDBBackend)
backends.register_backend("couchdbs", CouchDBBackend)
# Global registry for credentials for CouchDB Servers
_credentials_store: urllib.request.HTTPPasswordMgrWithPriorAuth = urllib.request.HTTPPasswordMgrWithPriorAuth()
_credentials_store: Dict[str, Tuple[str, str]] = {}
# Note: The HTTPPasswordMgr is not thread safe during writing, should be thread safe for reading only.
......@@ -178,7 +186,8 @@ def register_credentials(url: str, username: str, password: str):
:param username: Username to that CouchDB instance
:param password: Password to the Username
"""
_credentials_store.add_password(None, url, username, password, is_authenticated=True)
url_parts = urllib.parse.urlparse(url)
_credentials_store[url_parts.scheme + url_parts.netloc] = (username, password)
# Global registry for CouchDB Revisions
......@@ -253,12 +262,8 @@ class CouchDBObjectStore(model.AbstractObjectStore):
:param create: If True and the database does not exist, try to create it
:raises CouchDBError: If error occur during the request to the CouchDB server (see `_do_request()` for details)
"""
request = urllib.request.Request(
"{}/{}".format(self.url, self.database_name),
headers={'Accept': 'application/json'},
method='HEAD')
try:
CouchDBBackend.do_request(request)
CouchDBBackend.do_request("{}/{}".format(self.url, self.database_name), 'HEAD')
except CouchDBServerError as e:
# If an HTTPError is raised, re-raise it, unless it is a 404 error and we are requested to create the
# database
......@@ -270,11 +275,7 @@ class CouchDBObjectStore(model.AbstractObjectStore):
# Create database
logger.info("Creating CouchDB database %s/%s ...", self.url, self.database_name)
request = urllib.request.Request(
"{}/{}".format(self.url, self.database_name),
headers={'Accept': 'application/json'},
method='PUT')
CouchDBBackend.do_request(request)
CouchDBBackend.do_request("{}/{}".format(self.url, self.database_name), 'PUT')
def get_identifiable(self, identifier: Union[str, model.Identifier]) -> model.Identifiable:
"""
......@@ -290,11 +291,9 @@ class CouchDBObjectStore(model.AbstractObjectStore):
identifier = self._transform_id(identifier, False)
# Create and issue HTTP request (raises HTTPError on status != 200)
request = urllib.request.Request(
"{}/{}/{}".format(self.url, self.database_name, urllib.parse.quote(identifier, safe='')),
headers={'Accept': 'application/json'})
try:
data = CouchDBBackend.do_request(request)
data = CouchDBBackend.do_request(
"{}/{}/{}".format(self.url, self.database_name, urllib.parse.quote(identifier, safe='')))
except CouchDBServerError as e:
if e.code == 404:
raise KeyError("No Identifiable with id {} found in CouchDB database".format(identifier)) from e
......@@ -335,13 +334,12 @@ class CouchDBObjectStore(model.AbstractObjectStore):
data = json.dumps({'data': x}, cls=json_serialization.AASToJsonEncoder)
# Create and issue HTTP request (raises HTTPError on status != 200)
request = urllib.request.Request(
"{}/{}/{}".format(self.url, self.database_name, self._transform_id(x.identification)),
headers={'Content-type': 'application/json'},
method='PUT',
data=data.encode())
try:
response = CouchDBBackend.do_request(request)
response = CouchDBBackend.do_request(
"{}/{}/{}".format(self.url, self.database_name, self._transform_id(x.identification)),
'PUT',
{'Content-type': 'application/json'},
data.encode('utf-8'))
set_couchdb_revision("{}/{}/{}".format(self.url, self.database_name, self._transform_id(x.identification)),
response["rev"])
except CouchDBServerError as e:
......@@ -379,25 +377,19 @@ class CouchDBObjectStore(model.AbstractObjectStore):
# ETag response header
try:
logger.debug("fetching the current object revision for deletion ...")
request = urllib.request.Request(
"{}/{}/{}".format(self.url, self.database_name, self._transform_id(x.identification)),
headers={'Accept': 'application/json'},
method='HEAD')
opener = urllib.request.build_opener(urllib.request.HTTPBasicAuthHandler(_credentials_store))
response = opener.open(request)
rev = response.getheader('ETag')[1:-1]
except urllib.error.HTTPError as e:
headers = CouchDBBackend.do_request(
"{}/{}/{}".format(self.url, self.database_name, self._transform_id(x.identification)), 'HEAD')
rev = headers['ETag'][1:-1]
except CouchDBServerError as e:
if e.code == 404:
raise KeyError("No AAS object with id {} exists in CouchDB database".format(x.identification))\
from e
raise
request = urllib.request.Request(
"{}/{}/{}?rev={}".format(self.url, self.database_name, self._transform_id(x.identification), rev),
headers={'Content-type': 'application/json'},
method='DELETE')
try:
CouchDBBackend.do_request(request)
CouchDBBackend.do_request(
"{}/{}/{}?rev={}".format(self.url, self.database_name, self._transform_id(x.identification), rev),
'DELETE')
except CouchDBServerError as e:
if e.code == 404:
raise KeyError("No AAS object with id {} exists in CouchDB database".format(x.identification)) from e
......@@ -429,12 +421,9 @@ class CouchDBObjectStore(model.AbstractObjectStore):
else:
return False
logger.debug("Checking existence of object with id %s in database ...", repr(x))
request = urllib.request.Request(
"{}/{}/{}".format(self.url, self.database_name, self._transform_id(identifier)),
headers={'Accept': 'application/json'},
method='HEAD')
try:
CouchDBBackend.do_request(request)
CouchDBBackend.do_request(
"{}/{}/{}".format(self.url, self.database_name, self._transform_id(identifier)), 'HEAD')
except CouchDBServerError as e:
if e.code == 404:
return False
......@@ -449,10 +438,7 @@ class CouchDBObjectStore(model.AbstractObjectStore):
:raises CouchDBError: If error occur during the request to the CouchDB server (see `_do_request()` for details)
"""
logger.debug("Fetching number of documents from database ...")
request = urllib.request.Request(
"{}/{}".format(self.url, self.database_name),
headers={'Accept': 'application/json'})
data = CouchDBBackend.do_request(request)
data = CouchDBBackend.do_request("{}/{}".format(self.url, self.database_name))
return data['doc_count']
def __iter__(self) -> Iterator[model.Identifiable]:
......@@ -477,10 +463,7 @@ class CouchDBObjectStore(model.AbstractObjectStore):
# Fetch a list of all ids and construct Iterator object
logger.debug("Creating iterator over objects in database ...")
request = urllib.request.Request(
"{}/{}/_all_docs".format(self.url, self.database_name),
headers={'Accept': 'application/json'})
data = CouchDBBackend.do_request(request)
data = CouchDBBackend.do_request("{}/{}/_all_docs".format(self.url, self.database_name))
return CouchDBIdentifiableIterator(self, (row['id'] for row in data['rows']))
@staticmethod
......
aas/examples/tutorial_backend_couchdb.py 0 → 100755
View file @ 79230435
#!/usr/bin/env python3
# This work is licensed under a Creative Commons CCZero 1.0 Universal License.
# See http://creativecommons.org/publicdomain/zero/1.0/ for more information.
"""
Tutorial for storing Asset Administration Shells, Submodels and Assets in a CouchDB database server, using the
CouchDBObjectStore and CouchDB Backend.
This tutorial also shows the usage of the commit()/update() mechanism for synchronizing objects with an external data
source.
"""
from configparser import ConfigParser
from pathlib import Path
import aas.examples.data.example_aas
import aas.backend.couchdb
# To execute this tutorial, you'll need a running CouchDB server, including an empty database and a user account with
# access to that database.
# After installing CouchDB, you can use the CouchDB web interface "Fauxton" (typically at
# http://localhost:5984/_utils/) to create a new database. Optionally, you can create a new user by adding a document
# to the `_users` database with the following contents (notice that the username is required in two positions):
#
# {"_id": "org.couchdb.user:<your username>",
# "name": "<your username>",
# "password": "<your password>",
# "roles": [],
# "type": "user"}
#
# Afterwards you can add the new user to the set of "Members" of your new database (via the "Permissions" section in the
# user interface). Alternatively, you can use the admin credentials with PyI40AAS (see below).
# Step by Step Guide:
# step 1: connecting to a CouchDB server
# step 2: storing objects in the CouchDBObjectStore
# step 3: updating objects from the CouchDB and committing changes
##########################################
# Step 1: Connecting to a CouchDB Server #
##########################################
# Well, actually, connections to the CouchDB server are created by the CouchDB backend, as required. However, we need
# to provide the login credentials to the server for this to work.
#
# Here, we take the test configuration to work with PyI40AAS development environments. You should replace these with
# the url of your CouchDB server (typically http://localhost:5984), the name of the empty database, and the name and
# password of a CouchDB user account which is "member" of this database (see above). Alternatively, you can provide
# your CouchDB server's admin credentials.
config = ConfigParser()
config.read([Path(__file__).parent.parent.parent / 'test' / 'test_config.default.ini',
Path(__file__).parent.parent.parent / 'test' / 'test_config.ini'])
couchdb_url = config['couchdb']['url']
couchdb_database = config['couchdb']['database']
couchdb_user = config['couchdb']['user']
couchdb_password = config['couchdb']['password']
# Provide the login credentials to the CouchDB backend.
# These credetials are used, whenever communication with this CouchDB server is required (either via the
# CouchDBObjectStore or via the update()/commit() backend.
aas.backend.couchdb.register_credentials(couchdb_url, couchdb_user, couchdb_password)
# Now, we create a CouchDBObjectStore as an interface for managing the objects in the CouchDB server.
object_store = aas.backend.couchdb.CouchDBObjectStore(couchdb_url, couchdb_database)
#####################################################
# Step 2: Storing objects in the CouchDBObjectStore #
#####################################################
# Create some example objects
example_submodel1 = aas.examples.data.example_aas.create_example_asset_identification_submodel()
example_submodel2 = aas.examples.data.example_aas.create_example_bill_of_material_submodel()
# The CouchDBObjectStore behaves just like other ObjectStore implementations (see `tutorial_storage.py`). The objects
# are transferred to the CouchDB immediately. Additionally, the `source` attribute is set automatically, so update() and
# commit() will work automatically (see below).
object_store.add(example_submodel1)
object_store.add(example_submodel2)
###################################################################
# Step 3: Updating Objects from the CouchDB and Commiting Changes #
###################################################################
# Since the CouchDBObjectStore has set the `source` attribute of our Submodel objects, we can now use update() and
# commit() to synchronize changes to these objects with the database. The `source` indicates (via its URI scheme) that
# the CouchDB backend is used for the synchronization and references the correct CouchDB server url and database. For
# this to work, we must make sure to `import aas.backend.couchdb` at least once in this Python application, so the
# CouchDB backend is loaded.
# Fetch recent updates from the server
example_submodel1.update()
# Make some changes to a Property within the submodel
prop = example_submodel1.get_referable('ManufacturerName')
assert(isinstance(prop, aas.model.Property))
prop.value = "RWTH Aachen"
# Commit (upload) these changes to the CouchDB server
# We can simply call commit() on the Property object. It will check the `source` attribute of the object itself as well
# as the source attribute of all ancestors in the object hierarchy (including the Submodel) and commit the changes to
# all of these external data sources.
prop.commit()
############
# Clean up #
############
# Let's delete the Submodels from the CouchDB to leave it in a clean state
object_store.discard(example_submodel1)
object_store.discard(example_submodel2)
aas/examples/tutorial_storage.py
View file @ 79230435
......@@ -67,9 +67,10 @@ aas = AssetAdministrationShell(
# objects using a dict.
# This may not be a suitable solution, if you need to manage large numbers of objects or objects must kept in a
# persistent memory (i.e. on hard disk). In this case, you may chose the `CouchDBObjectStore` from
# `aas.adapter.couchdb` to use a CouchDB database server as persistent storage. Both ObjectStore implementations provide
# the same interface. Therefore, all the methods shown in this tutorial, can be realized with a CouchDBObjectStore as
# well.
# `aas.backends.couchdb` to use a CouchDB database server as persistent storage. Both ObjectStore implementations
# provide the same interface. In addition, the CouchDBObjectStores allows synchronizing the local object with the
# database via a Backend and the update()/commit() mechanism. See the `tutorial_backend_couchdb.py` for more
# information.
obj_store: model.DictObjectStore[model.Identifiable] = model.DictObjectStore()
# step 2.2: add asset, submodel and asset administration shell to store
......
requirements.txt
View file @ 79230435
......@@ -3,3 +3,4 @@ lxml>=4.2,<5
python-dateutil>=2.8,<3.0
pyecma376-2>=0.2.4
psutil>=5.6
urllib3>=1.26<2.0
setup.py
View file @ 79230435
......@@ -47,6 +47,7 @@ setuptools.setup(
install_requires=[
'python-dateutil>=2.8,<3',
'lxml>=4.2,<5',
'urllib3>=1.26<2.0',
'pyecma376-2>=0.2.4',
]
)
test/_helper/test_helpers.py 0 → 100644
View file @ 79230435
import configparser
import os.path
import urllib.request
import urllib.error
import base64
TEST_CONFIG = configparser.ConfigParser()
TEST_CONFIG.read((os.path.join(os.path.dirname(__file__), "..", "test_config.default.ini"),
os.path.join(os.path.dirname(__file__), "..", "test_config.ini")))
# Check if CouchDB database is available. Otherwise, skip tests.
try:
request = urllib.request.Request(
"{}/{}".format(TEST_CONFIG['couchdb']['url'], TEST_CONFIG['couchdb']['database']),
headers={
'Authorization': 'Basic %s' % base64.b64encode(
('%s:%s' % (TEST_CONFIG['couchdb']['user'], TEST_CONFIG['couchdb']['password']))
.encode('ascii')).decode("ascii")
},
method='HEAD')
urllib.request.urlopen(request)
COUCHDB_OKAY = True
COUCHDB_ERROR = None
except urllib.error.URLError as e:
COUCHDB_OKAY = False
COUCHDB_ERROR = e
test/backend/test_backends.py
View file @ 79230435
from unittest import mock
from typing import List
import unittest
from aas.backend import backends
from aas.model import Referable
class ExampleBackend(backends.Backend):
@classmethod
def commit_object(cls, committed_object: Referable, store_object: Referable, relative_path: List[str]) -> None:
raise NotImplementedError("This is a mock")
@classmethod
def update_object(cls, updated_object: Referable, store_object: Referable, relative_path: List[str]) -> None:
raise NotImplementedError("This is a mock")
class BackendsTest(unittest.TestCase):
def test_backend_store(self):
with mock.patch("aas.backend.backends.Backend") as mock_backend:
backends.register_backend("mockScheme", mock_backend)
self.assertEqual(backends.get_backend("mockScheme:x-test:test_backend"), mock_backend)
backends._backends_map = {}
backends.register_backend("<this is totally a valid uri>", mock_backend)
with self.assertRaises(ValueError) as cm:
backends.get_backend("<this is totally a valid uri>")
self.assertEqual("<this is totally a valid uri> is not a valid URL with URI scheme.", str(cm.exception))
backends.register_backend("mockScheme", ExampleBackend)
self.assertIs(backends.get_backend("mockScheme:x-test:test_backend"), ExampleBackend)
backends.register_backend("<this is totally a valid uri>", ExampleBackend)
with self.assertRaises(ValueError) as cm:
backends.get_backend("<this is totally a valid uri>")
self.assertEqual("<this is totally a valid uri> is not a valid URL with URI scheme.", str(cm.exception))
if __name__ == '__main__':
unittest.main()
with self.assertRaises(backends.UnknownBackendException):
backends.get_backend("some-unkown-scheme://example.com")
test/backend/test_couchdb.py
View file @ 79230435
......@@ -8,45 +8,20 @@
# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
import base64
import configparser
import copy
import os
import unittest
import unittest.mock
import urllib.request
import urllib.error
from aas.backend import backends, couchdb
from aas.backend import couchdb
from aas.examples.data.example_aas import *
from test._helper.test_helpers import TEST_CONFIG, COUCHDB_OKAY, COUCHDB_ERROR
TEST_CONFIG = configparser.ConfigParser()
TEST_CONFIG.read((os.path.join(os.path.dirname(__file__), "..", "test_config.default.ini"),
os.path.join(os.path.dirname(__file__), "..", "test_config.ini")))
source_core: str = "couchdb://" + TEST_CONFIG["couchdb"]["url"].lstrip("http://") + "/" + \
TEST_CONFIG["couchdb"]["database"] + "/"
# Check if CouchDB database is available. Otherwise, skip tests.