1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
/*
* ============LICENSE_START=======================================================
* Copyright (C) 2023 Nordix Foundation
* ================================================================================
* 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.
*
* SPDX-License-Identifier: Apache-2.0
* ============LICENSE_END=========================================================
*/
package org.onap.cps.integration.functional
import org.onap.cps.api.CpsAdminService
import org.onap.cps.integration.base.CpsIntegrationSpecBase
import org.onap.cps.spi.exceptions.AlreadyDefinedException
import org.onap.cps.spi.exceptions.AnchorNotFoundException
import org.onap.cps.spi.exceptions.DataspaceInUseException
import org.onap.cps.spi.exceptions.DataspaceNotFoundException
class CpsAdminServiceIntegrationSpec extends CpsIntegrationSpecBase {
CpsAdminService objectUnderTest
def setup() { objectUnderTest = cpsAdminService }
def 'Dataspace CRUD operations.'() {
when: 'a dataspace is created'
objectUnderTest.createDataspace('newDataspace')
then: 'the dataspace can be read'
assert objectUnderTest.getDataspace('newDataspace').name == 'newDataspace'
and: 'it can be deleted'
objectUnderTest.deleteDataspace('newDataspace')
then: 'the dataspace no longer exists i.e. an exception is thrown if an attempt is made to retrieve it'
def thrown = null
try {
objectUnderTest.getDataspace('newDataspace')
} catch(Exception e) {
thrown = e
}
assert thrown instanceof DataspaceNotFoundException
}
def 'Delete dataspace with error; #scenario.'() {
setup: 'add some anchors if needed'
numberOfAnchors.times {
objectUnderTest.createAnchor(GENERAL_TEST_DATASPACE, BOOKSTORE_SCHEMA_SET, 'anchor' + it)
}
when: 'attempt to delete dataspace'
objectUnderTest.deleteDataspace(dataspaceName)
then: 'the correct exception is thrown with the relevant details'
def thrownException = thrown(expectedException)
thrownException.details.contains(expectedMessageDetails)
cleanup:
numberOfAnchors.times {
objectUnderTest.deleteAnchor(GENERAL_TEST_DATASPACE, 'anchor' + it)
}
where: 'the following data is used'
scenario | dataspaceName | numberOfAnchors || expectedException | expectedMessageDetails
'dataspace name does not exist' | 'unknown' | 0 || DataspaceNotFoundException | 'unknown does not exist'
'dataspace contains schemasets' | GENERAL_TEST_DATASPACE | 0 || DataspaceInUseException | 'contains 1 schemaset(s)'
'dataspace contains anchors' | GENERAL_TEST_DATASPACE | 2 || DataspaceInUseException | 'contains 2 anchor(s)'
}
def 'Retrieve all dataspaces (depends on total test suite).'() {
given: 'two addtional dataspaces are created'
objectUnderTest.createDataspace('dataspace1')
objectUnderTest.createDataspace('dataspace2')
when: 'all datespaces are retreived'
def result = objectUnderTest.getAllDataspaces()
then: 'there are at least 3 dataspaces (2 new ones plus the general test dataspace)'
result.size() >= 3
assert result.name.containsAll([GENERAL_TEST_DATASPACE, 'dataspace1', 'dataspace2'])
}
def 'Duplicate dataspaces.'() {
when: 'attempting to create a dataspace with the same name as an existing one'
objectUnderTest.createDataspace(GENERAL_TEST_DATASPACE)
then: 'an exception is thrown indicating the dataspace already exists'
thrown(AlreadyDefinedException)
}
def 'Anchor CRUD operations.'() {
when: 'an anchor is created'
objectUnderTest.createAnchor(GENERAL_TEST_DATASPACE, BOOKSTORE_SCHEMA_SET, 'newAnchor')
then: 'the anchor be read'
assert objectUnderTest.getAnchor(GENERAL_TEST_DATASPACE, 'newAnchor').name == 'newAnchor'
and: 'it can be deleted'
objectUnderTest.deleteAnchor(GENERAL_TEST_DATASPACE,'newAnchor')
then: 'the anchor no longer exists i.e. an exception is thrown if an attempt is made to retrieve it'
def thrown = null
try {
objectUnderTest.getAnchor(GENERAL_TEST_DATASPACE, 'newAnchor')
} catch(Exception e) {
thrown = e
}
assert thrown instanceof AnchorNotFoundException
}
def 'Filtering multiple anchors.'() {
when: '2 anchors with bookstore schema set are created'
objectUnderTest.createAnchor(GENERAL_TEST_DATASPACE, BOOKSTORE_SCHEMA_SET, 'anchor1')
objectUnderTest.createAnchor(GENERAL_TEST_DATASPACE, BOOKSTORE_SCHEMA_SET, 'anchor2')
and: '1 anchor with "other" schema set is created'
def bookstoreModelFileContent = readResourceDataFile('bookstore/bookstore.yang')
cpsModuleService.createSchemaSet(GENERAL_TEST_DATASPACE, 'otherSchemaSet', [someFileName: bookstoreModelFileContent])
objectUnderTest.createAnchor(GENERAL_TEST_DATASPACE, 'otherSchemaSet', 'anchor3')
then: 'there are 3 anchors in the general test database'
assert objectUnderTest.getAnchors(GENERAL_TEST_DATASPACE).size() == 3
and: 'there are 2 anchors associated with bookstore schema set'
assert objectUnderTest.getAnchors(GENERAL_TEST_DATASPACE, BOOKSTORE_SCHEMA_SET).size() == 2
and: 'there is 1 anchor associated with other schema set'
assert objectUnderTest.getAnchors(GENERAL_TEST_DATASPACE, 'otherSchemaSet').size() == 1
}
def 'Querying anchor(name)s (depends on previous test!).'() {
expect: 'there are now 3 anchors using the "stores" module (both schema sets use the same modules) '
assert objectUnderTest.queryAnchorNames(GENERAL_TEST_DATASPACE, ['stores']).size() == 3
and: 'there are no anchors using both "stores" and a "unused-model"'
assert objectUnderTest.queryAnchorNames(GENERAL_TEST_DATASPACE, ['stores', 'unused-model']).size() == 0
}
def 'Duplicate anchors.'() {
given: 'an anchor is created'
objectUnderTest.createAnchor(GENERAL_TEST_DATASPACE, BOOKSTORE_SCHEMA_SET, 'newAnchor')
when: 'attempt to create another anchor with the same name'
objectUnderTest.createAnchor(GENERAL_TEST_DATASPACE, BOOKSTORE_SCHEMA_SET, 'newAnchor')
then: 'an exception is thrown that the anchor already is defined'
thrown(AlreadyDefinedException)
cleanup:
objectUnderTest.deleteAnchor(GENERAL_TEST_DATASPACE, 'newAnchor')
}
def 'Query anchors without any known modules and #scenario'() {
when: 'querying for anchors with #scenario'
def result = objectUnderTest.queryAnchorNames(dataspaceName, ['unknownModule'])
then: 'an empty result is returned (no error)'
assert result == []
where:
scenario | dataspaceName
'non existing database' | 'nonExistingDataspace'
'just unknown module(s)' | GENERAL_TEST_DATASPACE
}
}
|
