test_couchdb.py 9.6 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
# Copyright 2020 PyI40AAS Contributors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# 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.
11
import unittest
12
import unittest.mock
13
import urllib.error
14

Michael Thies's avatar
Michael Thies committed
15
from aas.backend import couchdb
16
from aas.examples.data.example_aas import *
17

18
from test._helper.test_helpers import TEST_CONFIG, COUCHDB_OKAY, COUCHDB_ERROR
19

20

21
source_core: str = "couchdb://" + TEST_CONFIG["couchdb"]["url"].lstrip("http://") + "/" + \
22
23
                   TEST_CONFIG["couchdb"]["database"] + "/"

24
25

class CouchDBBackendOfflineMethodsTest(unittest.TestCase):
26
    def test_parse_source(self):
27
        couchdb.register_credentials(url="couchdb.plt.rwth-aachen.de:5984",
28
29
30
                                     username="test_user",
                                     password="test_password")

31
        url = couchdb.CouchDBBackend._parse_source(
32
            "couchdbs://couchdb.plt.rwth-aachen.de:5984/path_to_db/path_to_doc"
33
34
35
36
        )
        expected_url = "https://couchdb.plt.rwth-aachen.de:5984/path_to_db/path_to_doc"
        self.assertEqual(expected_url, url)

37
        url = couchdb.CouchDBBackend._parse_source(
38
            "couchdb://couchdb.plt.rwth-aachen.de:5984/path_to_db/path_to_doc"
39
40
41
42
43
44
45
46
47
48
        )
        expected_url = "http://couchdb.plt.rwth-aachen.de:5984/path_to_db/path_to_doc"
        self.assertEqual(expected_url, url)

        with self.assertRaises(couchdb.CouchDBSourceError) as cm:
            couchdb.CouchDBBackend._parse_source("wrong_scheme:plt.rwth-aachen.couchdb:5984/path_to_db/path_to_doc")
            self.assertEqual("Source has wrong format. "
                             "Expected to start with {couchdb, couchdbs}, got "
                             "{wrong_scheme:plt.rwth-aachen.couchdb:5984/path_to_db/path_to_doc}",
                             cm.exception)
49
50
51
52
53
54


@unittest.skipUnless(COUCHDB_OKAY, "No CouchDB is reachable at {}/{}: {}".format(TEST_CONFIG['couchdb']['url'],
                                                                                 TEST_CONFIG['couchdb']['database'],
                                                                                 COUCHDB_ERROR))
class CouchDBBackendTest(unittest.TestCase):
55
56
57
58
    def setUp(self) -> None:
        self.object_store = couchdb.CouchDBObjectStore(TEST_CONFIG['couchdb']['url'],
                                                       TEST_CONFIG['couchdb']['database'])
        couchdb.register_credentials(TEST_CONFIG["couchdb"]["url"],
59
60
                                     TEST_CONFIG["couchdb"]["user"],
                                     TEST_CONFIG["couchdb"]["password"])
61
        self.object_store.check_database()
62

63
64
    def tearDown(self) -> None:
        self.object_store.clear()
65

66
    def test_object_store_add(self):
67
        test_object = create_example_submodel()
68
69
70
        self.object_store.add(test_object)
        self.assertEqual(test_object.source, source_core+"IRI-https%3A%2F%2Facplt.org%2FTest_Submodel")

71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
    def test_retrieval(self):
        test_object = create_example_submodel()
        self.object_store.add(test_object)

        # When retrieving the object, we should get the *same* instance as we added
        test_object_retrieved = self.object_store.get_identifiable(
            model.Identifier(id_='https://acplt.org/Test_Submodel', id_type=model.IdentifierType.IRI))
        self.assertIs(test_object, test_object_retrieved)

        # When retrieving it again, we should still get the same object
        del test_object
        test_object_retrieved_again = self.object_store.get_identifiable(
            model.Identifier(id_='https://acplt.org/Test_Submodel', id_type=model.IdentifierType.IRI))
        self.assertIs(test_object_retrieved, test_object_retrieved_again)

86
87
88
89
90
91
        # However, a changed source should invalidate the cached object, so we should get a new copy
        test_object_retrieved.source = "couchdb://example.com/example/IRI-https%3A%2F%2Facplt.org%2FTest_Submodel"
        test_object_retrieved_third = self.object_store.get_identifiable(
            model.Identifier(id_='https://acplt.org/Test_Submodel', id_type=model.IdentifierType.IRI))
        self.assertIsNot(test_object_retrieved, test_object_retrieved_third)

92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
    def test_example_submodel_storing(self) -> None:
        example_submodel = create_example_submodel()

        # Add exmaple submodel
        self.object_store.add(example_submodel)
        self.assertEqual(1, len(self.object_store))
        self.assertIn(example_submodel, self.object_store)

        # Restore example submodel and check data
        submodel_restored = self.object_store.get_identifiable(
            model.Identifier(id_='https://acplt.org/Test_Submodel', id_type=model.IdentifierType.IRI))
        assert (isinstance(submodel_restored, model.Submodel))
        checker = AASDataChecker(raise_immediately=True)
        check_example_submodel(checker, submodel_restored)

        # Delete example submodel
        self.object_store.discard(submodel_restored)
        self.assertNotIn(example_submodel, self.object_store)

    def test_iterating(self) -> None:
        example_data = create_full_example()

        # Add all objects
        for item in example_data:
            self.object_store.add(item)

        self.assertEqual(6, len(self.object_store))

        # Iterate objects, add them to a DictObjectStore and check them
        retrieved_data_store: model.provider.DictObjectStore[model.Identifiable] = model.provider.DictObjectStore()
        for item in self.object_store:
            retrieved_data_store.add(item)
        checker = AASDataChecker(raise_immediately=True)
        check_full_example(checker, retrieved_data_store)

    def test_key_errors(self) -> None:
        # Double adding an object should raise a KeyError
        example_submodel = create_example_submodel()
        self.object_store.add(example_submodel)
        with self.assertRaises(KeyError) as cm:
            self.object_store.add(example_submodel)
        self.assertEqual("'Identifiable with id Identifier(IRI=https://acplt.org/Test_Submodel) already exists in "
                         "CouchDB database'", str(cm.exception))

        # Querying a deleted object should raise a KeyError
        retrieved_submodel = self.object_store.get_identifiable(
            model.Identifier('https://acplt.org/Test_Submodel', model.IdentifierType.IRI))
        self.object_store.discard(example_submodel)
140
        with self.assertRaises(KeyError) as cm:
141
142
            self.object_store.get_identifiable(model.Identifier('https://acplt.org/Test_Submodel',
                                                                model.IdentifierType.IRI))
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
        self.assertEqual("'No Identifiable with id IRI-https://acplt.org/Test_Submodel found in CouchDB database'",
                         str(cm.exception))

        # Double deleting should also raise a KeyError
        with self.assertRaises(KeyError) as cm:
            self.object_store.discard(retrieved_submodel)
        self.assertEqual("'No AAS object with id Identifier(IRI=https://acplt.org/Test_Submodel) exists in "
                         "CouchDB database'", str(cm.exception))

    def test_conflict_errors(self):
        # Preperation: add object and retrieve it from the database
        example_submodel = create_example_submodel()
        self.object_store.add(example_submodel)
        retrieved_submodel = self.object_store.get_identifiable(
            model.Identifier('https://acplt.org/Test_Submodel', model.IdentifierType.IRI))

159
160
161
162
163
        # Simulate a concurrent modification (Commit submodel, while preventing that the couchdb revision store is
        # updated)
        with unittest.mock.patch("aas.backend.couchdb.set_couchdb_revision"):
            retrieved_submodel.commit()

164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
        # Committing changes to the retrieved object should now raise a conflict error
        retrieved_submodel.id_short = "myOtherNewIdShort"
        with self.assertRaises(couchdb.CouchDBConflictError) as cm:
            retrieved_submodel.commit()
        self.assertEqual("Could not commit changes to id Identifier(IRI=https://acplt.org/Test_Submodel) due to a "
                         "concurrent modification in the database.", str(cm.exception))

        # Deleting the submodel with safe_delete should also raise a conflict error. Deletion without safe_delete should
        # work
        with self.assertRaises(couchdb.CouchDBConflictError) as cm:
            self.object_store.discard(retrieved_submodel, True)
        self.assertEqual("Object with id Identifier(IRI=https://acplt.org/Test_Submodel) has been modified in the "
                         "database since the version requested to be deleted.", str(cm.exception))
        self.object_store.discard(retrieved_submodel, False)
        self.assertEqual(0, len(self.object_store))

180
181
        # Committing after deletion should not raise a conflict error due to removal of the source attribute
        retrieved_submodel.commit()
182
183
184
185
186
187
188
189

    def test_editing(self):
        test_object = create_example_submodel()
        self.object_store.add(test_object)

        # Test if commit uploads changes
        test_object.id_short = "SomeNewIdShort"
        test_object.commit()
190
191
192
193
194

        # Test if update restores changes
        test_object.id_short = "AnotherIdShort"
        test_object.update()
        self.assertEqual("SomeNewIdShort", test_object.id_short)