Add kafka listener for data updated events

See "Running via Docker Compose" section from README.md file to have an
example of event processing

Issue-ID: CPS-371
Signed-off-by: Bruno Sakoto <bruno.sakoto@bell.ca>
Change-Id: Id3abfa32fb04e07102a5f28e6e43a9b533391188

9 files changed, 508 insertions, 18 deletions

diff --git a/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerIntegrationSpec.groovy b/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerIntegrationSpec.groovy
new file mode 100644
index 0000000..4c362ad
--- /dev/null
+++ b/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerIntegrationSpec.groovy
@@ -0,0 +1,124 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (c) 2021 Bell Canada.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+*/
+
+package org.onap.cps.temporal.controller.event.listener.kafka
+
+import groovy.util.logging.Slf4j
+import org.onap.cps.event.model.CpsDataUpdatedEvent
+import org.onap.cps.temporal.repository.containers.TimescaleContainer
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.jdbc.core.JdbcTemplate
+import org.springframework.kafka.core.KafkaTemplate
+import org.springframework.test.context.DynamicPropertyRegistry
+import org.springframework.test.context.DynamicPropertySource
+import org.testcontainers.containers.KafkaContainer
+import spock.lang.Shared
+import spock.lang.Specification
+import spock.util.concurrent.PollingConditions
+
+import java.util.concurrent.TimeUnit
+
+/**
+ * Integration test specification for data updated event listener.
+ * This integration test is running database and kafka dependencies as docker containers.
+ */
+@SpringBootTest
+@Slf4j
+class DataUpdatedEventListenerIntegrationSpec extends Specification {
+
+    @Shared
+    def databaseTestContainer = TimescaleContainer.getInstance()
+
+    static kafkaTestContainer = new KafkaContainer()
+    static {
+        Runtime.getRuntime().addShutdownHook(new Thread(kafkaTestContainer::stop))
+    }
+
+    def setupSpec() {
+        databaseTestContainer.start()
+        kafkaTestContainer.start()
+    }
+
+    @Autowired
+    KafkaTemplate<String, CpsDataUpdatedEvent> kafkaTemplate
+
+    @Autowired
+    JdbcTemplate jdbcTemplate
+
+    @Value('${app.listener.data-updated.topic}')
+    String topic
+
+    // Define event data
+    def aTimestamp = EventFixtures.currentIsoTimestamp()
+    def aDataspace = 'my-dataspace'
+    def aSchemaSet = 'my-schema-set'
+    def anAnchor = 'my-anchor'
+
+    // Define sql queries for data validation
+    def sqlCount = "select count(*) from network_data"
+    def sqlSelect =  "select * from network_data"
+    def sqlWhereClause =
+            ' where observed_timestamp = to_timestamp(?, \'YYYY-MM-DD"T"HH24:MI:SS.USTZHTZM\') ' +
+                    'and dataspace = ? ' +
+                    'and schema_set = ? ' +
+                    'and anchor = ?'
+    def sqlCountWithConditions = sqlCount + sqlWhereClause
+    def sqlSelectWithConditions = sqlSelect + sqlWhereClause
+
+    def 'Processing a valid event'() {
+        given: "no event has been proceeded"
+            def initialRecordsCount =
+                    jdbcTemplate.queryForObject(sqlCountWithConditions, Integer.class,
+                            aTimestamp, aDataspace, aSchemaSet, anAnchor)
+            assert (initialRecordsCount == 0)
+        when: 'an event is produced'
+            def event =
+                    EventFixtures.buildEvent(
+                            timestamp: aTimestamp, dataspace: aDataspace, schemaSet: aSchemaSet, anchor: anAnchor)
+            this.kafkaTemplate.send(topic, event)
+        then: 'the event is proceeded'
+            def pollingCondition = new PollingConditions(timeout: 10, initialDelay: 1, factor: 2)
+            pollingCondition.eventually {
+                def finalRecordsCount =
+                        jdbcTemplate.queryForObject(
+                                sqlCountWithConditions, Integer.class, aTimestamp, aDataspace, aSchemaSet, anAnchor)
+                assert (finalRecordsCount == 1)
+            }
+            Map<String, Object> result =
+                    jdbcTemplate.queryForMap(sqlSelectWithConditions, aTimestamp, aDataspace, aSchemaSet, anAnchor)
+            log.debug("Data retrieved from db: {}", result)
+    }
+
+    def 'Processing an invalid event'() {
+        given: 'the number of network data records if known'
+            def initialRecordsCount = jdbcTemplate.queryForObject(sqlCount, Integer.class)
+        when: 'an invalid event is produced'
+            this.kafkaTemplate.send(topic, (CpsDataUpdatedEvent) null)
+        then: 'the event is not proceeded and no more network data record is created'
+            TimeUnit.SECONDS.sleep(3)
+            assert (jdbcTemplate.queryForObject(sqlCount, Integer.class) == initialRecordsCount)
+    }
+
+    @DynamicPropertySource
+    static void registerKafkaProperties(DynamicPropertyRegistry registry) {
+        registry.add("spring.kafka.bootstrap-servers", kafkaTestContainer::getBootstrapServers)
+    }
+
+}
diff --git a/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerSpec.groovy b/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerSpec.groovy
new file mode 100644
index 0000000..d3a407c
--- /dev/null
+++ b/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/DataUpdatedEventListenerSpec.groovy
@@ -0,0 +1,117 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (c) 2021 Bell Canada.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.temporal.controller.event.listener.kafka
+
+import org.mapstruct.factory.Mappers
+import org.onap.cps.event.model.CpsDataUpdatedEvent
+import org.onap.cps.temporal.controller.event.listener.exception.InvalidEventEnvelopException
+import org.onap.cps.temporal.controller.event.model.CpsDataUpdatedEventMapper
+import org.onap.cps.temporal.service.NetworkDataService
+import spock.lang.Specification
+
+import static org.onap.cps.temporal.controller.event.listener.exception.InvalidEventEnvelopException.InvalidField.ErrorType.MISSING
+import static org.onap.cps.temporal.controller.event.listener.exception.InvalidEventEnvelopException.InvalidField.ErrorType.UNEXPECTED
+
+/**
+ * Test specification for data updated event listener.
+ */
+class DataUpdatedEventListenerSpec extends Specification {
+
+    // Define event data
+    def anEventType = 'my-event-type'
+    def anEventSource = new URI('my-event-source')
+    def aTimestamp = EventFixtures.currentIsoTimestamp()
+    def aDataspace = 'my-dataspace'
+    def aSchemaSet = 'my-schema-set'
+    def anAnchor = 'my-anchor'
+    def aDataName = 'my-data-name'
+    def aDataValue = 'my-data-value'
+
+    // Define service mock
+    def mockService = Mock(NetworkDataService)
+
+    // Define mapper
+    def mapper = Mappers.getMapper(CpsDataUpdatedEventMapper.class)
+
+    // Define listener under test
+    def objectUnderTest = new DataUpdatedEventListener(mockService, mapper)
+
+    def 'Event message consumption'() {
+        when: 'an event is received'
+            def event =
+                    EventFixtures.buildEvent(
+                            timestamp: aTimestamp, dataspace: aDataspace, schemaSet: aSchemaSet, anchor: anAnchor,
+                            dataName: aDataName, dataValue: aDataValue)
+            objectUnderTest.consume(event)
+        then: 'network data service is requested to persisted the data change'
+            1 * mockService.addNetworkData(
+                    {
+                        it.getObservedTimestamp() == EventFixtures.toOffsetDateTime(aTimestamp)
+                        && it.getDataspace() == aDataspace
+                        && it.getSchemaSet() == aSchemaSet
+                        && it.getAnchor() == anAnchor
+                        && it.getPayload() == String.format('{"%s":"%s"}', aDataName, aDataValue)
+                        && it.getCreatedTimestamp() == null
+                    }
+            )
+    }
+
+    def 'Event message consumption fails because of missing envelop'() {
+        when: 'an event without envelop information is received'
+            def invalidEvent = new CpsDataUpdatedEvent().withSchema(null)
+            objectUnderTest.consume(invalidEvent)
+        then: 'an exception is thrown with 4 invalid fields'
+            def e = thrown(InvalidEventEnvelopException)
+            e.getInvalidFields().size() == 4
+            e.getInvalidFields().contains(
+                    new InvalidEventEnvelopException.InvalidField(
+                            MISSING,"schema", null,
+                            CpsDataUpdatedEvent.Schema.URN_CPS_ORG_ONAP_CPS_DATA_UPDATED_EVENT_SCHEMA_1_1_0_SNAPSHOT
+                                    .value()))
+            e.getInvalidFields().contains(
+                    new InvalidEventEnvelopException.InvalidField(
+                            MISSING, "id", null, null))
+            e.getInvalidFields().contains(
+                    new InvalidEventEnvelopException.InvalidField(
+                            UNEXPECTED, "source", null, EventFixtures.defaultEventSource.toString()))
+            e.getInvalidFields().contains(
+                    new InvalidEventEnvelopException.InvalidField(
+                            UNEXPECTED, "type", null, EventFixtures.defaultEventType))
+            e.getMessage().contains(e.getInvalidFields().toString())
+    }
+
+    def 'Event message consumption fails because of invalid envelop'() {
+        when: 'an event with an invalid envelop is received'
+            def invalidEvent =
+                    new CpsDataUpdatedEvent()
+                            .withId('my-id').withSource(anEventSource).withType(anEventType)
+            objectUnderTest.consume(invalidEvent)
+        then: 'an exception is thrown with 2 invalid fields'
+            def e = thrown(InvalidEventEnvelopException)
+            e.getInvalidFields().size() == 2
+            e.getInvalidFields().contains(
+                    new InvalidEventEnvelopException.InvalidField(
+                            UNEXPECTED, "type", anEventType, EventFixtures.defaultEventType))
+            e.getInvalidFields().contains(
+                    new InvalidEventEnvelopException.InvalidField(
+                            UNEXPECTED, "source", anEventSource.toString(),
+                            EventFixtures.defaultEventSource.toString()))
+    }
+
+}
diff --git a/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/EventFixtures.groovy b/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/EventFixtures.groovy
new file mode 100644
index 0000000..44a28de
--- /dev/null
+++ b/src/test/groovy/org/onap/cps/temporal/controller/event/listener/kafka/EventFixtures.groovy
@@ -0,0 +1,72 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (c) 2021 Bell Canada.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.temporal.controller.event.listener.kafka
+
+import org.onap.cps.event.model.Content
+import org.onap.cps.event.model.CpsDataUpdatedEvent
+import org.onap.cps.event.model.Data
+
+import java.time.OffsetDateTime
+import java.time.format.DateTimeFormatter
+
+/**
+ * This class contains utility fixtures methods for building and manipulating event data.
+ */
+class EventFixtures {
+
+    static DateTimeFormatter isoTimestampFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ")
+    static String defaultEventType = 'org.onap.cps.data-updated-event'
+    static URI defaultEventSource = new URI('urn:cps:org.onap.cps')
+
+    static CpsDataUpdatedEvent buildEvent(final Map map) {
+        CpsDataUpdatedEvent event =
+                new CpsDataUpdatedEvent()
+                        .withId(
+                                map.id != null ? map.id.toString() : UUID.randomUUID().toString())
+                        .withType(
+                                map.eventType != null ? map.eventType.toString() : defaultEventType)
+                        .withSource(
+                                map.eventSource != null ? new URI(map.eventSource.toString()) : defaultEventSource)
+                        .withContent(
+                                new Content()
+                                        .withObservedTimestamp(
+                                                map.timestamp != null ? map.timestamp.toString() : currentTimestamp())
+                                        .withDataspaceName(
+                                                map.dataspace != null ? map.dataspace.toString() : 'a-dataspace')
+                                        .withSchemaSetName(
+                                                map.schemaSet != null ? map.schemaSet.toString() : 'a-schema-set')
+                                        .withAnchorName(
+                                                map.anchor != null ? map.anchor.toString() : 'an-anchor')
+                                        .withData(
+                                                new Data().withAdditionalProperty(
+                                                        map.dataName != null ? map.dataName.toString() : 'a-data-name',
+                                                        map.dataValue != null ? map.dataValue : 'a-data-value')))
+
+        return event
+    }
+
+    static String currentIsoTimestamp() {
+        return isoTimestampFormatter.format(OffsetDateTime.now())
+    }
+
+    static OffsetDateTime toOffsetDateTime(String timestamp) {
+        return OffsetDateTime.parse(timestamp, isoTimestampFormatter)
+    }
+
+}
diff --git a/src/test/groovy/org/onap/cps/temporal/controller/event/model/CpsDataUpdatedEventMapperSpec.groovy b/src/test/groovy/org/onap/cps/temporal/controller/event/model/CpsDataUpdatedEventMapperSpec.groovy
new file mode 100644
index 0000000..132ff6d
--- /dev/null
+++ b/src/test/groovy/org/onap/cps/temporal/controller/event/model/CpsDataUpdatedEventMapperSpec.groovy
@@ -0,0 +1,131 @@
+/*
+ * ============LICENSE_START=======================================================
+ * Copyright (c) 2021 Bell Canada.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.cps.temporal.controller.event.model
+
+import com.fasterxml.jackson.core.JsonProcessingException
+import org.mapstruct.factory.Mappers
+import org.onap.cps.event.model.Content
+import org.onap.cps.event.model.CpsDataUpdatedEvent
+import org.onap.cps.event.model.Data
+import org.onap.cps.temporal.domain.NetworkData
+import spock.lang.Specification
+
+import java.time.OffsetDateTime
+import java.time.format.DateTimeFormatter
+
+/**
+ * Test specification for data updated event mapper.
+ */
+class CpsDataUpdatedEventMapperSpec extends Specification {
+
+    def objectUnderTest = Mappers.getMapper(CpsDataUpdatedEventMapper.class);
+
+    def 'Mapping a null event'() {
+        given: 'a null event'
+            def event = null
+        when: 'the event is mapped to an entity'
+            NetworkData result = objectUnderTest.eventToEntity(event)
+        then: 'the result entity is null'
+            result == null
+    }
+
+    def 'Mapping an event whose properties are null'() {
+        given: 'an event whose properties are null'
+            def event = new CpsDataUpdatedEvent()
+        when: 'the event is mapped to an entity'
+            NetworkData result = objectUnderTest.eventToEntity(event)
+        then: 'the result entity is not null'
+            result != null
+        and: 'all result entity properties are null'
+            assertEntityPropertiesAreNull(result)
+    }
+
+    def 'Mapping an event whose content properties are null'() {
+        given: 'an event whose content properties are null'
+            def event = new CpsDataUpdatedEvent().withContent(new Content())
+        when: 'the event is mapped to an entity'
+            NetworkData result = objectUnderTest.eventToEntity(event)
+        then: 'the result entity is not null'
+            result != null
+        and: 'all result entity properties are null'
+            assertEntityPropertiesAreNull(result)
+    }
+
+    def 'Mapping an event whose content data is empty'() {
+        given: 'an event whose content data is empty'
+            def event = new CpsDataUpdatedEvent().withContent(new Content().withData(new Data()))
+        when: 'the event is mapped to an entity'
+            NetworkData result = objectUnderTest.eventToEntity(event)
+        then: 'the result entity is not null'
+            result != null
+        and: 'the result entity payload is an empty json '
+            result.getPayload() == "{}"
+    }
+
+    def 'Mapping an event whose content data is invalid'() {
+        given: 'an event whose content data is invalid'
+            def event =
+                    new CpsDataUpdatedEvent().withContent(new Content().withData(
+                            new Data().withAdditionalProperty(null, null)))
+        when: 'the event is mapped to an entity'
+            NetworkData result = objectUnderTest.eventToEntity(event)
+        then: 'an runtime exception is thrown'
+            def e = thrown(RuntimeException)
+            e.getCause() instanceof JsonProcessingException
+    }
+
+    def 'Mapping a valid complete event'() {
+        given: 'a valid complete event'
+            def isoTimestampFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSZ")
+            def aDataName = 'a-data-name'
+            def aDataValue = 'a-data-value'
+            def event =
+                    new CpsDataUpdatedEvent()
+                            .withContent(
+                                    new Content()
+                                            .withObservedTimestamp(isoTimestampFormatter.format(OffsetDateTime.now()))
+                                            .withDataspaceName('a-dataspace')
+                                            .withSchemaSetName('a-schema-set')
+                                            .withAnchorName('an-anchor')
+                                            .withData(new Data().withAdditionalProperty(aDataName, aDataValue)))
+        when: 'the event is mapped to an entity'
+            NetworkData result = objectUnderTest.eventToEntity(event)
+        then: 'the result entity is not null'
+            result != null
+        and: 'all result entity properties are the ones from the event'
+            result.getObservedTimestamp() ==
+                    OffsetDateTime.parse(event.getContent().getObservedTimestamp(), isoTimestampFormatter)
+            result.getDataspace() == event.getContent().getDataspaceName()
+            result.getSchemaSet() == event.getContent().getSchemaSetName()
+            result.getAnchor() == event.getContent().getAnchorName()
+            result.getPayload().contains(aDataValue)
+            result.getPayload().contains(aDataValue)
+            result.getCreatedTimestamp() == null
+    }
+
+    private void assertEntityPropertiesAreNull(NetworkData networkData) {
+        assert networkData.getObservedTimestamp() == null
+        assert networkData.getDataspace() == null
+        assert networkData.getSchemaSet() == null
+        assert networkData.getAnchor() == null
+        assert networkData.getPayload() == null
+        assert networkData.getCreatedTimestamp() == null
+    }
+
+}
diff --git a/src/test/groovy/org/onap/cps/temporal/controller/QueryControllerSpec.groovy b/src/test/groovy/org/onap/cps/temporal/controller/web/QueryControllerSpec.groovy
index f718bf4..c834f38 100644
--- a/src/test/groovy/org/onap/cps/temporal/controller/QueryControllerSpec.groovy
+++ b/src/test/groovy/org/onap/cps/temporal/controller/web/QueryControllerSpec.groovy
@@ -16,7 +16,7 @@
  * ============LICENSE_END=========================================================
  */
 
-package org.onap.cps.temporal.controller
+package org.onap.cps.temporal.controller.web
 
 import spock.lang.Specification
 
@@ -34,4 +34,4 @@ class QueryControllerSpec extends Specification {
             ! response.empty
     }
 
-}
\ No newline at end of file
+}
diff --git a/src/test/groovy/org/onap/cps/temporal/repository/NetworkDataRepositorySpec.groovy b/src/test/groovy/org/onap/cps/temporal/repository/NetworkDataRepositorySpec.groovy
index ec976ee..f66b35e 100644
--- a/src/test/groovy/org/onap/cps/temporal/repository/NetworkDataRepositorySpec.groovy
+++ b/src/test/groovy/org/onap/cps/temporal/repository/NetworkDataRepositorySpec.groovy
@@ -29,6 +29,9 @@ import spock.lang.Specification
 
 import java.time.OffsetDateTime
 
+/**
+ * Test specification for network data repository.
+ */
 @SpringBootTest
 @Testcontainers
 class NetworkDataRepositorySpec extends Specification {
diff --git a/src/test/groovy/org/onap/cps/temporal/service/NetworkDataServiceImplSpec.groovy b/src/test/groovy/org/onap/cps/temporal/service/NetworkDataServiceImplSpec.groovy
index 70ac2bc..9847f54 100644
--- a/src/test/groovy/org/onap/cps/temporal/service/NetworkDataServiceImplSpec.groovy
+++ b/src/test/groovy/org/onap/cps/temporal/service/NetworkDataServiceImplSpec.groovy
@@ -18,28 +18,51 @@
 
 package org.onap.cps.temporal.service
 
+import org.onap.cps.temporal.domain.NetworkDataId
+
 import java.time.OffsetDateTime
 import org.onap.cps.temporal.domain.NetworkData
 import org.onap.cps.temporal.repository.NetworkDataRepository
 import spock.lang.Specification
 
+/**
+ * Test specification for network data service.
+ */
 class NetworkDataServiceImplSpec extends Specification {
 
-    def objectUnderTest = new NetworkDataServiceImpl()
-
     def mockNetworkDataRepository = Mock(NetworkDataRepository)
 
+    def objectUnderTest = new NetworkDataServiceImpl(mockNetworkDataRepository)
+
     def networkData = new NetworkData()
 
-    def setup() {
-        objectUnderTest.networkDataRepository = mockNetworkDataRepository
+    def 'Add network data successfully.'() {
+        given: 'network data repository is persisting network data it is asked to save'
+            def persistedNetworkData = new NetworkData()
+            persistedNetworkData.setCreatedTimestamp(OffsetDateTime.now())
+            mockNetworkDataRepository.save(networkData) >> persistedNetworkData
+        when: 'a new network data is added'
+            def result = objectUnderTest.addNetworkData(networkData)
+        then: 'result network data is the one that has been persisted'
+            result == persistedNetworkData
+            result.getCreatedTimestamp() != null
+            networkData.getCreatedTimestamp() == null
     }
 
-    def 'Add network data in timeseries database.'() {
+    def 'Add network data fails because already added'() {
+        given: 'network data repository is not able to create data it is asked to persist ' +
+                'and reveals it with null created timestamp on network data entity'
+            def persistedNetworkData = new NetworkData()
+            persistedNetworkData.setCreatedTimestamp(null)
+            mockNetworkDataRepository.save(networkData) >> persistedNetworkData
+        and: 'existing data can be retrieved'
+            def existing = new NetworkData()
+            existing.setCreatedTimestamp(OffsetDateTime.now().minusYears(1))
+            mockNetworkDataRepository.findById(_ as NetworkDataId) >> Optional.of(existing)
         when: 'a new network data is added'
             objectUnderTest.addNetworkData(networkData)
-        then: ' repository service is called with the correct parameters'
-            1 * mockNetworkDataRepository.save(networkData)
+        then: 'network service exception is thrown'
+            thrown(ServiceException)
     }
 
 }
diff --git a/src/test/java/org/onap/cps/temporal/architecture/LayeredArchitectureTest.java b/src/test/java/org/onap/cps/temporal/architecture/LayeredArchitectureTest.java
index d47e8a5..a70e914 100644
--- a/src/test/java/org/onap/cps/temporal/architecture/LayeredArchitectureTest.java
+++ b/src/test/java/org/onap/cps/temporal/architecture/LayeredArchitectureTest.java
@@ -70,4 +70,4 @@ public class LayeredArchitectureTest {
                     .should().onlyHaveDependentClassesThat()
                     .resideInAnyPackage(SERVICE_PACKAGE, REPOSITORY_PACKAGE);
 
-}
\ No newline at end of file
+}
diff --git a/src/test/resources/application.yml b/src/test/resources/application.yml
index afaff6c..3ac13a9 100644
--- a/src/test/resources/application.yml
+++ b/src/test/resources/application.yml
@@ -20,17 +20,37 @@ server:
 spring:
     datasource:
         url: ${DB_URL}
-        password: ${DB_PASSWORD}
         username: ${DB_USERNAME}
+        password: ${DB_PASSWORD}
     liquibase:
         change-log: classpath:/db/changelog/changelog-master.xml
     jpa:
-        open-in-view: false
         properties:
-            hibernate:
-                dialect: org.hibernate.dialect.PostgreSQLDialect
+            hibernate.dialect: org.hibernate.dialect.PostgreSQLDialect
+            hibernate.format_sql: true
+            hibernate.generate_statistics: false
+    kafka:
+        bootstrap-servers: localhost:9092
+        security:
+            protocol: PLAINTEXT
+        consumer:
+            group-id: cps-temporal-group
+            # Configures the Spring Kafka ErrorHandlingDeserializer that delegates to the 'real' deserializers
+            # See https://docs.spring.io/spring-kafka/docs/2.5.11.RELEASE/reference/html/#error-handling-deserializer
+            # and https://www.confluent.io/blog/spring-kafka-can-your-kafka-consumers-handle-a-poison-pill/
+            key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
+            value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
+            auto-offset-reset: earliest
+            properties:
+                spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
+                spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
+                spring.json.value.default.type: org.onap.cps.event.model.CpsDataUpdatedEvent
+        # Following is not cps-temporal configuration. It is configuration for the producer used for integration tests
+        producer:
+            key-serializer: org.apache.kafka.common.serialization.StringSerializer
+            value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
 
-logging:
-    level:
-        org:
-            springframework: INFO
+app:
+    listener:
+        data-updated:
+            topic: cps.cfg-state-events